DATAKV-22

Initial draft support for Redis pubsub
+ introduce PubSub contract
+ add adapters for Jedis (no-op for JRedis which does not support PubSub)
+ introduce dedicated exception for subscribed connections
+ add low-level MessageListener and Subscription mechanisms

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java

@@ -0,0 +1,47 @@
+/*
+ * 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;
+
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+
+/**
+ * Exception thrown when issuing commands on a connection that is subscribed and waiting
+ * for events.
+ * 
+ * @author Costin Leau
+ * @see RedisConnection#subscribe(org.springframework.data.keyvalue.redis.connection.MessageListener, byte[]...)
+ */
+public class SubscribedRedisConnectionException extends InvalidDataAccessApiUsageException {
+
+	/**
+	 * Constructs a new <code>SubscribedRedisConnectionException</code> instance.
+	 *
+	 * @param msg
+	 * @param cause
+	 */
+	public SubscribedRedisConnectionException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+
+	/**
+	 * Constructs a new <code>SubscribedRedisConnectionException</code> instance.
+	 *
+	 * @param msg
+	 */
+	public SubscribedRedisConnectionException(String msg) {
+		super(msg);
+	}
+}

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

@@ -0,0 +1,33 @@
+/*
+ * 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;
+
+/**
+ * Listener of messages published in Redis.
+ * 
+ * @author Costin Leau
+ */
+public interface MessageListener {
+
+	/**
+	 * Callback for processing received objects through Redis.
+	 * 
+	 * @param message message
+	 * @param channel Redis channel
+	 * @param pattern channel pattern - matching pattern (if used), null otherwise.
+	 */
+	void onMessage(byte[] message, byte[] channel, byte[] pattern);
+}

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

@@ -25,7 +25,7 @@ import java.util.List;
  * @author Costin Leau
  */
 public interface RedisCommands extends RedisTxCommands, RedisStringCommands, RedisListCommands, RedisSetCommands,
-		RedisZSetCommands, RedisHashCommands, RedisServerCommands {
+		RedisZSetCommands, RedisHashCommands, RedisServerCommands, RedisPubSubCommands {
 
 	Boolean exists(byte[] key);
 

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

@@ -0,0 +1,79 @@
+/*
+ * 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;
+
+/**
+ * PubSub-specific Redis commands.
+ * 
+ * @author Costin Leau
+ */
+public interface RedisPubSubCommands {
+
+	/**
+	 * Indicates whether the current connection is subscribed (to at least one channel)
+	 * or not.
+	 * 
+	 * @return true if the connection is subscribed, false otherwise
+	 * @see #subscribe(Subscription, byte[]...)
+	 */
+	boolean isSubscribed();
+
+	/**
+	 * Returns the current subscription for this connection or null if the connection is
+	 * not subscribed.
+	 * 
+	 * @return the current subscription, null if none is available
+	 * @see #subscribe(Subscription, byte[]...)
+	 */
+	Subscription getSubscription();
+
+	/**
+	 * Publishes the given message to the given channel.
+	 * 
+	 * @param message message to publish
+	 * @param channel the channel to publish to
+	 * @return the number of clients that received the message
+	 */
+	Long publish(byte[] message, byte[] channel);
+
+	/**
+	 * Subscribes the connection to the given channels.
+	 * Once subscribed, a connection
+	 * enters listening mode and can only subscribe to other channels or unsubscribe.
+	 * No other commands are accepted until the connection is unsubscribed.
+	 * <p/>
+	 * Note that this operation is blocking and the current thread starts waiting
+	 * for new messages immediately.
+	 * 
+	 * @param subscription message subscription
+	 * @param channels channel names
+	 */
+	void subscribe(MessageListener listener, byte[]... channels);
+
+	/**
+	 * Subscribes the connection to all channels matching the given patterns.
+	 * Once subscribed, a connection
+	 * enters listening mode and can only subscribe to other channels or unsubscribe.
+	 * No other commands are accepted until the connection is unsubscribed.
+	 * <p/>
+	 * Note that this operation is blocking and the current thread starts waiting
+	 * for new messages immediately.
+	 * 
+	 * @param subscription message subscription
+	 * @param patterns channel name patterns
+	 */
+	void pSubscribe(MessageListener listener, byte[]... patterns);
+}

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

@@ -0,0 +1,93 @@
+/*
+ * 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;
+
+import java.util.Collection;
+
+/**
+ * Subscription for Redis channels.
+ * 
+ * @author Costin Leau
+ */
+public interface Subscription {
+
+	/**
+	 * Adds the given channels to the current subscription.
+	 * 
+	 * @param channels channel names
+	 */
+	void subscribe(byte[]... channels);
+
+	/**
+	 * Adds the given channel patterns to the current subscription.
+	 * 
+	 * @param patterns channel patterns
+	 */
+	void pSubscribe(byte[]... patterns);
+
+	/**
+	 * Cancels the current subscription for all channels given by name.
+	 */
+	void unsubscribe();
+
+	/**
+	 * Cancels the current subscription for all given channels.
+	 * 
+	 * @param channels channel names
+	 */
+	void unsubscribe(byte[]... channels);
+
+	/**
+	 * Cancels the subscription for all channels matched by patterns.
+	 */
+	void pUnsubscribe();
+
+	/**
+	 * Cancels the subscription for all channels matching the given patterns.
+	 *  
+	 * @param patterns
+	 */
+	void pUnsubscribe(byte[]... patterns);
+
+	/**
+	 * Returns the (named) channels for this subscription.
+	 * 
+	 * @return collection of named channels
+	 */
+	Collection<byte[]> getChannels();
+
+	/**
+	 * Returns the channel patters for this subscription.
+	 * 
+	 * @return collection of channel patterns
+	 */
+	Collection<byte[]> getPatterns();
+
+	/**
+	 * Returns the listener used for this subscription.
+	 * 
+	 * @return the listener used for this subscription.
+	 */
+	MessageListener getListener();
+
+	/**
+	 * Indicates whether this subscription is still 'alive'
+	 * or not. 
+	 * 
+	 * @return true if the subscription still applies, false otherwise.
+	 */
+	boolean isAlive();
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java

@@ -26,10 +26,13 @@ import java.util.Set;
 
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException;
+import org.springframework.data.keyvalue.redis.SubscribedRedisConnectionException;
 import org.springframework.data.keyvalue.redis.UncategorizedRedisException;
 import org.springframework.data.keyvalue.redis.connection.DataType;
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
 import org.springframework.data.keyvalue.redis.connection.SortParameters;
+import org.springframework.data.keyvalue.redis.connection.Subscription;
 import org.springframework.util.ReflectionUtils;
 
 import redis.clients.jedis.BinaryJedis;
@@ -37,6 +40,7 @@ import redis.clients.jedis.BinaryTransaction;
 import redis.clients.jedis.Client;
 import redis.clients.jedis.Jedis;
 import redis.clients.jedis.JedisException;
+import redis.clients.jedis.JedisPubSub;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.Transaction;
 import redis.clients.jedis.ZParams;
@@ -59,6 +63,8 @@ public class JedisConnection implements RedisConnection {
 	private final Client client;
 	private final BinaryTransaction transaction;
 
+	private volatile JedisSubscription subscription;
+
 	/**
 	 * Constructs a new <code>JedisConnection</code> instance.
 	 *
@@ -1567,4 +1573,84 @@ public class JedisConnection implements RedisConnection {
 			throw convertJedisAccessException(ex);
 		}
 	}
+
+
+	//
+	// Pub/Sub functionality
+	//
+	@Override
+	public Long publish(byte[] message, byte[] channel) {
+		try {
+			if (isQueueing()) {
+				throw new UnsupportedOperationException();
+			}
+
+			String msg = new String(message);
+			String chn = new String(channel);
+
+			return jedis.publish(chn, msg);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	@Override
+	public Subscription getSubscription() {
+		return subscription;
+	}
+
+	@Override
+	public boolean isSubscribed() {
+		return (subscription != null && subscription.isAlive());
+	}
+
+	@Override
+	public void pSubscribe(MessageListener listener, byte[]... patterns) {
+		if (isSubscribed()) {
+			throw new SubscribedRedisConnectionException(
+					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
+		}
+
+		try {
+			if (isQueueing()) {
+				throw new UnsupportedOperationException();
+			}
+
+			String[] pats = JedisUtils.convert(patterns);
+			JedisPubSub jedisPubSub = JedisUtils.adaptPubSub(listener);
+
+			subscription = new JedisSubscription(listener, jedisPubSub);
+			jedis.psubscribe(jedisPubSub, pats);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	@Override
+	public void subscribe(MessageListener listener, byte[]... channels) {
+		if (isSubscribed()) {
+			throw new SubscribedRedisConnectionException(
+					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
+		}
+
+		try {
+			if (isQueueing()) {
+				throw new UnsupportedOperationException();
+			}
+
+			String[] chs = JedisUtils.convert(channels);
+			JedisPubSub jedisPubSub = JedisUtils.adaptPubSub(listener);
+
+			subscription = new JedisSubscription(listener, jedisPubSub);
+			jedis.subscribe(jedisPubSub, chs);
+		} catch (Exception ex) {
+			throw convertJedisAccessException(ex);
+		}
+	}
+
+	private void checkSubscription() {
+		if (isSubscribed()) {
+			throw new SubscribedRedisConnectionException("Cannot execute command - connection is subscribed");
+		}
+	}
 }

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisMessageListener.java

@@ -0,0 +1,66 @@
+/*
+ * 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.jedis;
+
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
+import org.springframework.util.Assert;
+
+import redis.clients.jedis.JedisPubSub;
+
+/**
+ * MessageListener adapter on top of Jedis.
+ * 
+ * @author Costin Leau
+ */
+class JedisMessageListener extends JedisPubSub {
+
+	private final MessageListener listener;
+
+	JedisMessageListener(MessageListener listener) {
+		Assert.notNull(listener, "message listener is required");
+		this.listener = listener;
+	}
+
+	@Override
+	public void onMessage(String channel, String message) {
+		listener.onMessage(message.getBytes(), channel.getBytes(), null);
+	}
+
+	@Override
+	public void onPMessage(String pattern, String channel, String message) {
+		listener.onMessage(message.getBytes(), channel.getBytes(), pattern.getBytes());
+	}
+
+	@Override
+	public void onPSubscribe(String pattern, int subscribedChannels) {
+		// no-op
+	}
+
+	@Override
+	public void onPUnsubscribe(String pattern, int subscribedChannels) {
+		// no-op	
+	}
+
+	@Override
+	public void onSubscribe(String channel, int subscribedChannels) {
+		// no-op
+	}
+
+	@Override
+	public void onUnsubscribe(String channel, int subscribedChannels) {
+		// no-op	
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java

@@ -0,0 +1,126 @@
+/*
+ * 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.jedis;
+
+import java.util.ArrayList;
+import java.util.Collection;
+
+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;
+
+import redis.clients.jedis.JedisPubSub;
+
+/**
+ * Jedis specific subscription.
+ * 
+ * @author Costin Leau
+ */
+class JedisSubscription implements Subscription {
+
+	private final MessageListener listener;
+	private final JedisPubSub jedisPubSub;
+
+	private final Collection<byte[]> channels = new ArrayList<byte[]>(2);
+	private final Collection<byte[]> patterns = new ArrayList<byte[]>(2);
+
+	JedisSubscription(MessageListener listener, JedisPubSub jedisPubSub) {
+		Assert.notNull(listener);
+		this.listener = listener;
+	}
+
+	@Override
+	public Collection<byte[]> getChannels() {
+		return channels;
+	}
+
+	@Override
+	public MessageListener getListener() {
+		return listener;
+	}
+
+	@Override
+	public Collection<byte[]> getPatterns() {
+		return patterns;
+	}
+
+	@Override
+	public void pSubscribe(byte[]... patterns) {
+		Assert.notEmpty(patterns, "at least one pattern required");
+
+		for (byte[] bs : patterns) {
+			this.patterns.add(bs);
+		}
+
+		jedisPubSub.psubscribe(JedisUtils.convert(patterns));
+	}
+
+	@Override
+	public void pUnsubscribe() {
+		jedisPubSub.punsubscribe();
+	}
+
+	@Override
+	public void pUnsubscribe(byte[]... patterns) {
+		if (ObjectUtils.isEmpty(patterns)) {
+			unsubscribe();
+		}
+
+		else {
+			for (byte[] bs : patterns) {
+				this.patterns.remove(bs);
+			}
+
+			jedisPubSub.punsubscribe(JedisUtils.convert(patterns));
+		}
+	}
+
+	@Override
+	public void subscribe(byte[]... channels) {
+		Assert.notEmpty(patterns, "at least one pattern required");
+
+		for (byte[] bs : patterns) {
+			this.patterns.add(bs);
+		}
+
+		jedisPubSub.subscribe(JedisUtils.convert(channels));
+	}
+
+	@Override
+	public void unsubscribe() {
+		jedisPubSub.unsubscribe();
+	}
+
+	@Override
+	public void unsubscribe(byte[]... channels) {
+		if (ObjectUtils.isEmpty(channels)) {
+			unsubscribe();
+		}
+		else {
+			for (byte[] bs : patterns) {
+				this.patterns.remove(bs);
+			}
+
+			jedisPubSub.unsubscribe(JedisUtils.convert(channels));
+		}
+	}
+
+	@Override
+	public boolean isAlive() {
+		return jedisPubSub.isSubscribed();
+	}
+}

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2010 the original author or authors.
+ * Copyright 2010-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.
@@ -31,6 +31,7 @@ import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.data.keyvalue.redis.RedisConnectionFailureException;
 import org.springframework.data.keyvalue.redis.UncategorizedRedisException;
 import org.springframework.data.keyvalue.redis.connection.DefaultTuple;
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
 import org.springframework.data.keyvalue.redis.connection.SortParameters;
 import org.springframework.data.keyvalue.redis.connection.RedisListCommands.POSITION;
 import org.springframework.data.keyvalue.redis.connection.RedisZSetCommands.Tuple;
@@ -39,6 +40,7 @@ import org.springframework.data.keyvalue.redis.connection.SortParameters.Range;
 import org.springframework.util.Assert;
 
 import redis.clients.jedis.JedisException;
+import redis.clients.jedis.JedisPubSub;
 import redis.clients.jedis.SortingParams;
 import redis.clients.jedis.BinaryClient.LIST_POSITION;
 
@@ -189,4 +191,18 @@ public abstract class JedisUtils {
 		}
 		return info;
 	}
+
+	static JedisPubSub adaptPubSub(MessageListener listener) {
+		return new JedisMessageListener(listener);
+	}
+
+	static String[] convert(byte[]... raw) {
+		String[] result = new String[raw.length];
+
+		for (int i = 0; i < raw.length; i++) {
+			result[i] = new String(raw[i]);
+		}
+
+		return result;
+	}
 }

spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java

@@ -31,8 +31,10 @@ import org.springframework.dao.DataAccessException;
 import org.springframework.data.keyvalue.UncategorizedKeyvalueStoreException;
 import org.springframework.data.keyvalue.redis.UncategorizedRedisException;
 import org.springframework.data.keyvalue.redis.connection.DataType;
+import org.springframework.data.keyvalue.redis.connection.MessageListener;
 import org.springframework.data.keyvalue.redis.connection.RedisConnection;
 import org.springframework.data.keyvalue.redis.connection.SortParameters;
+import org.springframework.data.keyvalue.redis.connection.Subscription;
 
 /**
  * {@code RedisConnection} implementation on top of <a href="http://github.com/alphazero/jredis">JRedis</a> library.
@@ -1025,4 +1027,33 @@ public class JredisConnection implements RedisConnection {
 			throw JredisUtils.convertJredisAccessException(ex);
 		}
 	}
+
+	//
+	// PubSub commands
+	//
+
+	@Override
+	public Subscription getSubscription() {
+		return null;
+	}
+
+	@Override
+	public boolean isSubscribed() {
+		return false;
+	}
+
+	@Override
+	public void pSubscribe(MessageListener listener, byte[]... patterns) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public Long publish(byte[] message, byte[] channel) {
+		throw new UnsupportedOperationException();
+	}
+
+	@Override
+	public void subscribe(MessageListener listener, byte[]... channels) {
+		throw new UnsupportedOperationException();
+	}
 }