DATAKV-46

+ wrap up RJC connector with pub sub support

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java

@@ -27,6 +27,7 @@ import org.idevlab.rjc.Session;
 import org.idevlab.rjc.SessionFactoryImpl;
 import org.idevlab.rjc.SortingParams;
 import org.idevlab.rjc.ZParams;
+import org.idevlab.rjc.message.RedisNodeSubscriber;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException;
 import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException;
@@ -50,9 +51,14 @@ public class RjcConnection implements RedisConnection {
 	private final Session session;
 	private volatile Client pipeline;
 
+	private volatile RjcSubscription subscription;
+	private volatile RedisNodeSubscriber subscriber;
+
 	public RjcConnection(org.idevlab.rjc.ds.RedisConnection connection, int dbIndex) {
-		session = new SessionFactoryImpl(new SingleDataSource(connection)).create();
+		SingleDataSource connectionDataSource = new SingleDataSource(connection);
+		session = new SessionFactoryImpl().create();
 		client = new Client(connection);
+		subscriber = new RedisNodeSubscriber(connectionDataSource);
 
 		this.dbIndex = dbIndex;
 
@@ -73,6 +79,7 @@ public class RjcConnection implements RedisConnection {
 	public void close() throws DataAccessException {
 		isClosed = true;
 		try {
+			subscriber.close();
 			session.close();
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
@@ -1969,7 +1976,7 @@ public class RjcConnection implements RedisConnection {
 			if (isPipelined()) {
 				throw new UnsupportedOperationException();
 			}
-			return session.publish(channel, message);
+			return session.publish(RjcUtils.decode(channel), RjcUtils.decode(message));
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
 		}
@@ -1987,8 +1994,6 @@ public class RjcConnection implements RedisConnection {
 
 	@Override
 	public void pSubscribe(MessageListener listener, byte[]... patterns) {
-		String[] stringKeys = RjcUtils.decodeMultiple(patterns);
-
 		if (isSubscribed()) {
 			throw new SubscribedRedisConnectionException(
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
@@ -2002,10 +2007,9 @@ public class RjcConnection implements RedisConnection {
 				throw new UnsupportedOperationException();
 			}
 
-			BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener);
+			subscription = new RjcSubscription(listener, subscriber);
+			subscription.pSubscribe(patterns);
 
-			subscription = new sessionSubscription(listener, sessionPubSub, null, patterns);
-			session.psubscribe(sessionPubSub, patterns);
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
 		}
@@ -2013,8 +2017,6 @@ public class RjcConnection implements RedisConnection {
 
 	@Override
 	public void subscribe(MessageListener listener, byte[]... channels) {
-		String[] stringKeys = RjcUtils.decodeMultiple(channels);
-
 		if (isSubscribed()) {
 			throw new SubscribedRedisConnectionException(
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
@@ -2028,10 +2030,9 @@ public class RjcConnection implements RedisConnection {
 				throw new UnsupportedOperationException();
 			}
 
-			BinarysessionPubSub sessionPubSub = RjcUtils.adaptPubSub(listener);
+			subscription = new RjcSubscription(listener, subscriber);
+			subscription.pSubscribe(channels);
 
-			subscription = new sessionSubscription(listener, sessionPubSub, channels, null);
-			session.subscribe(sessionPubSub, channels);
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
 		}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java

@@ -0,0 +1,45 @@
+/*
+ * Copyright 2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.keyvalue.redis.connection.rjc;
+
+import org.idevlab.rjc.message.MessageListener;
+import org.idevlab.rjc.message.PMessageListener;
+import org.springframework.data.keyvalue.redis.connection.DefaultMessage;
+
+/**
+ * Message listener adapter for RJC library.
+ * 
+ * @author Costin Leau
+ */
+class RjcMessageListener implements MessageListener, PMessageListener {
+
+	private final org.springframework.data.keyvalue.redis.connection.MessageListener listener;
+
+	RjcMessageListener(org.springframework.data.keyvalue.redis.connection.MessageListener messageListener) {
+		this.listener = messageListener;
+	}
+
+	@Override
+	public void onMessage(String channel, String message) {
+		listener.onMessage(new DefaultMessage(RjcUtils.encode(channel), RjcUtils.encode(message)), null);
+	}
+
+	@Override
+	public void onMessage(String pattern, String channel, String message) {
+		listener.onMessage(new DefaultMessage(RjcUtils.encode(channel), RjcUtils.encode(message)),
+				RjcUtils.encode(pattern));
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java

@@ -0,0 +1,149 @@
+/*
+ * Copyright 2011 the original author or authors.
+ * 
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ * 
+ *      http://www.apache.org/licenses/LICENSE-2.0
+ * 
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.keyvalue.redis.connection.rjc;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+import org.idevlab.rjc.message.RedisSubscriber;
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
+import org.springframework.data.keyvalue.redis.connection.Subscription;
+import org.springframework.util.Assert;
+import org.springframework.util.ObjectUtils;
+
+/**
+ * Message subscription on top of RJC.
+ * 
+ * @author Costin Leau
+ */
+class RjcSubscription implements Subscription {
+
+	private final MessageListener listener;
+	private final RedisSubscriber subscriber;
+	private final RjcMessageListener listenerAdapter;
+
+	private final Collection<byte[]> channels = new ArrayList<byte[]>(2);
+	private final Collection<byte[]> patterns = new ArrayList<byte[]>(2);
+
+	RjcSubscription(MessageListener listener, RedisSubscriber subscriber) {
+		Assert.notNull(listener);
+		this.listener = listener;
+		this.subscriber = subscriber;
+		this.listenerAdapter = new RjcMessageListener(listener);
+	}
+
+	@Override
+	public Collection<byte[]> getChannels() {
+		synchronized (channels) {
+			return new ArrayList<byte[]>(channels);
+		}
+	}
+
+	@Override
+	public MessageListener getListener() {
+		return listener;
+	}
+
+	@Override
+	public Collection<byte[]> getPatterns() {
+		synchronized (patterns) {
+			return new ArrayList<byte[]>(patterns);
+		}
+	}
+
+	@Override
+	public void pSubscribe(byte[]... patterns) {
+		Assert.notEmpty(patterns, "at least one pattern required");
+
+		synchronized (this.patterns) {
+			for (byte[] bs : patterns) {
+				this.patterns.add(bs);
+			}
+		}
+
+		for (String pattern : RjcUtils.decodeMultiple(patterns)) {
+			subscriber.psubscribe(pattern, listenerAdapter);
+		}
+	}
+
+	@Override
+	public void pUnsubscribe() {
+		pUnsubscribe(null);
+
+		synchronized (patterns) {
+			patterns.clear();
+		}
+	}
+
+	@Override
+	public void pUnsubscribe(byte[]... patterns) {
+		if (ObjectUtils.isEmpty(patterns)) {
+			patterns = this.patterns.toArray(new byte[this.patterns.size()][]);
+		}
+
+		synchronized (this.patterns) {
+			for (byte[] bs : patterns) {
+				this.patterns.remove(bs);
+			}
+		}
+
+		subscriber.punsubscribe(RjcUtils.decodeMultiple(patterns));
+	}
+
+	@Override
+	public void subscribe(byte[]... channels) {
+		Assert.notEmpty(channels, "at least one channel required");
+
+		synchronized (this.channels) {
+			for (byte[] bs : channels) {
+				this.channels.add(bs);
+			}
+		}
+
+		for (String channel : RjcUtils.decodeMultiple(channels)) {
+			subscriber.subscribe(channel, listenerAdapter);
+		}
+	}
+
+	@Override
+	public void unsubscribe() {
+		unsubscribe(null);
+
+		synchronized (patterns) {
+			patterns.clear();
+		}
+	}
+
+	@Override
+	public void unsubscribe(byte[]... channels) {
+		if (ObjectUtils.isEmpty(channels)) {
+			channels = this.channels.toArray(new byte[this.channels.size()][]);
+		}
+
+		synchronized (this.channels) {
+			for (byte[] bs : channels) {
+				this.channels.remove(bs);
+			}
+		}
+
+		subscriber.punsubscribe(RjcUtils.decodeMultiple(channels));
+	}
+
+	@Override
+	public boolean isAlive() {
+		return (!channels.isEmpty() || !patterns.isEmpty());
+	}
+}