From 6cf1a58606344d52eedfafe280ab5f9323e98e7e Mon Sep 17 00:00:00 2001 From: Costin Leau Date: Tue, 5 Apr 2011 21:51:13 +0300 Subject: [PATCH] Revert "+ upgrade to Rjc 0.6.4 (snapshot for now)" This reverts commit db0522429aa9c25c0f808b19adf973de054c9237. + Downgrading RJC dependency to 0.6.3 since 0.6.4 is not yet released --- spring-data-keyvalue-parent/pom.xml | 4 +- spring-data-redis/pom.xml | 4 +- .../rjc/CloseSuppressingRjcConnection.java | 125 ------------------ .../redis/connection/rjc/RjcConnection.java | 14 +- .../redis/connection/rjc/RjcSubscription.java | 39 +++++- 5 files changed, 45 insertions(+), 141 deletions(-) delete mode 100644 spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/CloseSuppressingRjcConnection.java diff --git a/spring-data-keyvalue-parent/pom.xml b/spring-data-keyvalue-parent/pom.xml index 45e6f69bd..b0a7cece1 100644 --- a/spring-data-keyvalue-parent/pom.xml +++ b/spring-data-keyvalue-parent/pom.xml @@ -314,8 +314,8 @@ org.apache.maven.plugins maven-compiler-plugin - 1.5 - 1.5 + 1.6 + 1.6 -Xlint:all true false diff --git a/spring-data-redis/pom.xml b/spring-data-redis/pom.xml index d67f0874d..2c95966af 100644 --- a/spring-data-redis/pom.xml +++ b/spring-data-redis/pom.xml @@ -16,10 +16,10 @@ "[3.0.0, 4.0.0)" 03122010 1.5.2 - 0.6.4-SNAPSHOT + 0.6.3 "[1.0.0,2.0.0)" "[1.6, 2.0.0)" - "[0.6.4, 0.6.4]" + "[0.6.3, 0.6.3]" diff --git a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/CloseSuppressingRjcConnection.java b/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/CloseSuppressingRjcConnection.java deleted file mode 100644 index c902cdf98..000000000 --- a/spring-data-redis/src/main/java/org/springframework/data/keyvalue/redis/connection/rjc/CloseSuppressingRjcConnection.java +++ /dev/null @@ -1,125 +0,0 @@ -/* - * 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.io.IOException; -import java.net.UnknownHostException; -import java.util.List; - -import org.idevlab.rjc.ds.RedisConnection; -import org.idevlab.rjc.message.RedisNodeSubscriber; -import org.idevlab.rjc.protocol.Protocol.Command; - -/** - * Basic decorator suppressing close() calls to the underlying connection. - * Used for reusing arbitrary connections with {@link RedisNodeSubscriber} without - * resorting to connection pooling. - * - * @author Costin Leau - */ -class CloseSuppressingRjcConnection implements RedisConnection { - - private final RedisConnection delegate; - - /** - * Constructs a new CloseSuppressingRjcConnection instance. - * - * @param delegate - */ - CloseSuppressingRjcConnection(RedisConnection delegate) { - this.delegate = delegate; - } - - public void close() { - // no-op - } - - public void connect() throws UnknownHostException, IOException { - delegate.connect(); - } - - public List getAll() { - return delegate.getAll(); - } - - public byte[] getBinaryBulkReply() { - return delegate.getBinaryBulkReply(); - } - - public List getBinaryMultiBulkReply() { - return delegate.getBinaryMultiBulkReply(); - } - - public String getBulkReply() { - return delegate.getBulkReply(); - } - - public String getHost() { - return delegate.getHost(); - } - - public Long getIntegerReply() { - return delegate.getIntegerReply(); - } - - public List getMultiBulkReply() { - return delegate.getMultiBulkReply(); - } - - public List getObjectMultiBulkReply() { - return delegate.getObjectMultiBulkReply(); - } - - public Object getOne() { - return delegate.getOne(); - } - - public int getPort() { - return delegate.getPort(); - } - - public String getStatusCodeReply() { - return delegate.getStatusCodeReply(); - } - - public int getTimeout() { - return delegate.getTimeout(); - } - - public boolean isConnected() { - return delegate.isConnected(); - } - - public void rollbackTimeout() { - delegate.rollbackTimeout(); - } - - public void sendCommand(Command arg0, byte[]... arg1) { - delegate.sendCommand(arg0, arg1); - } - - public void sendCommand(Command arg0, String... arg1) { - delegate.sendCommand(arg0, arg1); - } - - public void sendCommand(Command arg0) { - delegate.sendCommand(arg0); - } - - public void setTimeoutInfinite() { - delegate.setTimeoutInfinite(); - } -} \ No newline at end of file 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 50d5f78f0..a52af61d5 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 @@ -58,7 +58,7 @@ public class RjcConnection implements RedisConnection { SingleDataSource connectionDataSource = new SingleDataSource(connection); session = new SessionFactoryImpl(connectionDataSource).create(); subscriber = new RedisNodeSubscriber(); - subscriber.setDataSource(new SingleDataSource(new CloseSuppressingRjcConnection(connection))); + subscriber.setDataSource(connectionDataSource); client = new Client(connection); this.dbIndex = dbIndex; @@ -79,9 +79,13 @@ public class RjcConnection implements RedisConnection { @Override public void close() throws DataAccessException { isClosed = true; - try { subscriber.close(); + } catch (Exception ex) { + // ignore + } + + try { session.close(); } catch (Exception ex) { throw convertRjcAccessException(ex); @@ -2024,9 +2028,8 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new RjcSubscription(listener, subscriber); + subscription = new RjcSubscription(listener, subscriber, client); subscription.pSubscribe(patterns); - subscriber.runSubscription(); } catch (Exception ex) { throw convertRjcAccessException(ex); @@ -2048,9 +2051,8 @@ public class RjcConnection implements RedisConnection { throw new UnsupportedOperationException(); } - subscription = new RjcSubscription(listener, subscriber); + subscription = new RjcSubscription(listener, subscriber, client); subscription.subscribe(channels); - subscriber.runSubscription(); } catch (Exception ex) { throw convertRjcAccessException(ex); 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 index 78c5f2277..64d8bc959 100644 --- 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 @@ -15,6 +15,7 @@ */ package org.springframework.data.keyvalue.redis.connection.rjc; +import org.idevlab.rjc.Client; import org.idevlab.rjc.message.RedisNodeSubscriber; import org.springframework.data.keyvalue.redis.connection.MessageListener; import org.springframework.data.keyvalue.redis.connection.util.AbstractSubscription; @@ -27,36 +28,62 @@ import org.springframework.data.keyvalue.redis.connection.util.AbstractSubscript class RjcSubscription extends AbstractSubscription { private final RedisNodeSubscriber subscriber; + private final Client client; + // rjc does not support subscription while listening + // so we have to handle this ourselves through the client + private volatile boolean subscribed = false; - RjcSubscription(MessageListener listener, RedisNodeSubscriber subscriber) { + RjcSubscription(MessageListener listener, RedisNodeSubscriber subscriber, Client client) { super(listener); this.subscriber = subscriber; subscriber.setMessageListener(new RjcMessageListener(listener)); subscriber.setPMessageListener(new RjcMessageListener(listener)); + this.client = client; } @Override protected void doClose() { - subscriber.close(); + subscribed = false; + client.unsubscribe(); + client.punsubscribe(); + client.rollbackTimeout(); } @Override protected void doPsubscribe(byte[]... patterns) { - subscriber.psubscribe(RjcUtils.decodeMultiple(patterns)); + String[] pats = RjcUtils.decodeMultiple(patterns); + + if (subscribed) { + client.psubscribe(pats); + } + else { + subscriber.setPatterns(RjcUtils.addArray(subscriber.getPatterns(), pats)); + subscribed = true; + subscriber.subscribe(); + } } @Override protected void doPUnsubscribe(boolean all, byte[]... patterns) { - subscriber.punsubscribe(RjcUtils.decodeMultiple(patterns)); + client.punsubscribe(RjcUtils.decodeMultiple(patterns)); } @Override protected void doSubscribe(byte[]... channels) { - subscriber.subscribe(RjcUtils.decodeMultiple(channels)); + String[] chs = RjcUtils.decodeMultiple(channels); + + if (subscribed) { + client.subscribe(chs); + } + else { + subscriber.setPatterns(RjcUtils.addArray(subscriber.getChannels(), chs)); + subscribed = true; + subscriber.subscribe(); + } } @Override protected void doUnsubscribe(boolean all, byte[]... channels) { - subscriber.punsubscribe(RjcUtils.decodeMultiple(channels)); + client.punsubscribe(RjcUtils.decodeMultiple(channels)); } } \ No newline at end of file