diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java index a95e5a2d3..93e452427 100644 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcConnection.java +++ b/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); } diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcMessageListener.java new file mode 100644 index 000000000..c16a2040f --- /dev/null +++ b/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)); + } +} \ No newline at end of file diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/RjcSubscription.java new file mode 100644 index 000000000..a1075a120 --- /dev/null +++ b/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 channels = new ArrayList(2); + private final Collection patterns = new ArrayList(2); + + RjcSubscription(MessageListener listener, RedisSubscriber subscriber) { + Assert.notNull(listener); + this.listener = listener; + this.subscriber = subscriber; + this.listenerAdapter = new RjcMessageListener(listener); + } + + @Override + public Collection getChannels() { + synchronized (channels) { + return new ArrayList(channels); + } + } + + @Override + public MessageListener getListener() { + return listener; + } + + @Override + public Collection getPatterns() { + synchronized (patterns) { + return new ArrayList(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()); + } +} \ No newline at end of file