diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/SubscribedRedisConnectionException.java new file mode 100644 index 000000000..980105de1 --- /dev/null +++ b/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 SubscribedRedisConnectionException instance. + * + * @param msg + * @param cause + */ + public SubscribedRedisConnectionException(String msg, Throwable cause) { + super(msg, cause); + } + + /** + * Constructs a new SubscribedRedisConnectionException instance. + * + * @param msg + */ + public SubscribedRedisConnectionException(String msg) { + super(msg); + } +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/MessageListener.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/MessageListener.java new file mode 100644 index 000000000..538706789 --- /dev/null +++ b/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); +} diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisCommands.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisCommands.java index 2df2b6bc5..31a989954 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisCommands.java +++ b/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); diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisPubSubCommands.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/RedisPubSubCommands.java new file mode 100644 index 000000000..1d99737be --- /dev/null +++ b/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. + *

+ * 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. + *

+ * 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); +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/Subscription.java new file mode 100644 index 000000000..3000820f1 --- /dev/null +++ b/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 getChannels(); + + /** + * Returns the channel patters for this subscription. + * + * @return collection of channel patterns + */ + Collection 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(); +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java index dd45b9b94..dc7069f47 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisConnection.java +++ b/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 JedisConnection 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"); + } + } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisMessageListener.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisMessageListener.java new file mode 100644 index 000000000..318de09a1 --- /dev/null +++ b/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 + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisSubscription.java new file mode 100644 index 000000000..a185d2d42 --- /dev/null +++ b/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 channels = new ArrayList(2); + private final Collection patterns = new ArrayList(2); + + JedisSubscription(MessageListener listener, JedisPubSub jedisPubSub) { + Assert.notNull(listener); + this.listener = listener; + } + + @Override + public Collection getChannels() { + return channels; + } + + @Override + public MessageListener getListener() { + return listener; + } + + @Override + public Collection 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(); + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java index fb16f3f4a..e4327c1ce 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jedis/JedisUtils.java +++ b/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; + } } \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java index a20158e04..94325176f 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/jredis/JredisConnection.java +++ b/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 JRedis 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(); + } } \ No newline at end of file