Revert "Revert "+ upgrade to Rjc 0.6.4 (snapshot for now)""

This reverts commit 6cf1a58606.

Upgrade back to RJC 0.6.4 now that it has been released (just in time for M3)

spring-data-keyvalue-parent/pom.xml

@@ -314,8 +314,8 @@
         <groupId>org.apache.maven.plugins</groupId>
         <artifactId>maven-compiler-plugin</artifactId>
         <configuration>
-          <source>1.6</source>
-          <target>1.6</target>
+          <source>1.5</source>
+          <target>1.5</target>
           <compilerArgument>-Xlint:all</compilerArgument>
           <showWarnings>true</showWarnings>
           <showDeprecation>false</showDeprecation>

spring-data-redis/pom.xml

@@ -16,10 +16,10 @@
     <spring.range>"[3.0.0, 4.0.0)"</spring.range>
     <jredis.ver>03122010</jredis.ver>
     <jedis.ver>1.5.2</jedis.ver>
-    <rjc.ver>0.6.3</rjc.ver>
+    <rjc.ver>0.6.4</rjc.ver>
     <jedis.range>"[1.0.0,2.0.0)"</jedis.range>
     <jackson.range>"[1.6, 2.0.0)"</jackson.range>
-    <rjc.range>"[0.6.3, 0.6.3]"</rjc.range>
+    <rjc.range>"[0.6.4, 0.6.4]"</rjc.range>
 	</properties>
 
 	<dependencies>

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

@@ -0,0 +1,117 @@
+/*
+ * 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 <code>CloseSuppressingRjcConnection</code> instance.
+	 *
+	 * @param delegate
+	 */
+	CloseSuppressingRjcConnection(RedisConnection delegate) {
+		this.delegate = delegate;
+	}
+
+	public void close() {
+		// no-op
+	}
+
+	public void connect() throws UnknownHostException, IOException {
+		delegate.connect();
+	}
+
+	public List<Object> getAll() {
+		return delegate.getAll();
+	}
+
+	public String getBulkReply() {
+		return delegate.getBulkReply();
+	}
+
+	public String getHost() {
+		return delegate.getHost();
+	}
+
+	public Long getIntegerReply() {
+		return delegate.getIntegerReply();
+	}
+
+	public List<String> getMultiBulkReply() {
+		return delegate.getMultiBulkReply();
+	}
+
+	public List<Object> 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();
+	}
+}

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(connectionDataSource);
+		subscriber.setDataSource(new SingleDataSource(new CloseSuppressingRjcConnection(connection)));
 		client = new Client(connection);
 
 		this.dbIndex = dbIndex;
@@ -79,13 +79,9 @@ public class RjcConnection implements RedisConnection {
 	@Override
 	public void close() throws DataAccessException {
 		isClosed = true;
-		try {
-			subscriber.close();
-		} catch (Exception ex) {
-			// ignore
-		}
 
 		try {
+			subscriber.close();
 			session.close();
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
@@ -2028,8 +2024,9 @@ public class RjcConnection implements RedisConnection {
 				throw new UnsupportedOperationException();
 			}
 
-			subscription = new RjcSubscription(listener, subscriber, client);
+			subscription = new RjcSubscription(listener, subscriber);
 			subscription.pSubscribe(patterns);
+			subscriber.runSubscription();
 
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);
@@ -2051,8 +2048,9 @@ public class RjcConnection implements RedisConnection {
 				throw new UnsupportedOperationException();
 			}
 
-			subscription = new RjcSubscription(listener, subscriber, client);
+			subscription = new RjcSubscription(listener, subscriber);
 			subscription.subscribe(channels);
+			subscriber.runSubscription();
 
 		} catch (Exception ex) {
 			throw convertRjcAccessException(ex);

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

@@ -15,7 +15,6 @@
  */
 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;
@@ -28,62 +27,36 @@ 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, Client client) {
+	RjcSubscription(MessageListener listener, RedisNodeSubscriber subscriber) {
 		super(listener);
 		this.subscriber = subscriber;
 		subscriber.setMessageListener(new RjcMessageListener(listener));
 		subscriber.setPMessageListener(new RjcMessageListener(listener));
-		this.client = client;
 	}
 
 	@Override
 	protected void doClose() {
-		subscribed = false;
-		client.unsubscribe();
-		client.punsubscribe();
-		client.rollbackTimeout();
+		subscriber.close();
 	}
 
 	@Override
 	protected void doPsubscribe(byte[]... patterns) {
-		String[] pats = RjcUtils.decodeMultiple(patterns);
-		
-		if (subscribed) {
-			client.psubscribe(pats);
-		}
-		else {
-			subscriber.setPatterns(RjcUtils.addArray(subscriber.getPatterns(), pats));
-			subscribed = true;
-			subscriber.subscribe();
-		}
+		subscriber.psubscribe(RjcUtils.decodeMultiple(patterns));
 	}
 
 	@Override
 	protected void doPUnsubscribe(boolean all, byte[]... patterns) {
-		client.punsubscribe(RjcUtils.decodeMultiple(patterns));
+		subscriber.punsubscribe(RjcUtils.decodeMultiple(patterns));
 	}
 
 	@Override
 	protected void doSubscribe(byte[]... channels) {
-		String[] chs = RjcUtils.decodeMultiple(channels);
-
-		if (subscribed) {
-			client.subscribe(chs);
-		}
-		else {
-			subscriber.setPatterns(RjcUtils.addArray(subscriber.getChannels(), chs));
-			subscribed = true;
-			subscriber.subscribe();
-		}
+		subscriber.subscribe(RjcUtils.decodeMultiple(channels));
 	}
 
 	@Override
 	protected void doUnsubscribe(boolean all, byte[]... channels) {
-		client.punsubscribe(RjcUtils.decodeMultiple(channels));
+		subscriber.punsubscribe(RjcUtils.decodeMultiple(channels));
 	}
 }