Fix UnifiedJedis watch transaction state

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

@@ -68,6 +68,7 @@
  * @author Dengliming
  * @author John Blum
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  * @see redis.clients.jedis.Jedis
  * @see redis.clients.jedis.RedisClient
  */
@@ -211,6 +212,14 @@ private static DefaultJedisClientConfig createConfig(int dbIndex, @Nullable Stri
 
 		return doWithJedis(it -> {
 
+			if (delegate.isWatchOnly()) {
+
+				Response<Object> response = pipelineFunction.apply(JedisInvoker.createCommands(getRequiredTransaction()));
+				Object result = response.get();
+
+				return result != null ? converter.convert(result) : nullDefault.get();
+			}
+
 			if (isQueueing()) {
 
 				Response<Object> response = pipelineFunction.apply(JedisInvoker.createCommands(getRequiredTransaction()));
@@ -310,15 +319,17 @@ public Object execute(@NonNull String command, byte @NonNull []... args) {
 
 			ProtocolCommand protocolCommand = () -> JedisConverters.toBytes(command);
 
-			if (isQueueing() || isPipelined()) {
+			if (isQueueing() || isPipelined() || isWatchOnly()) {
 
 				CommandArguments arguments = new CommandArguments(protocolCommand).addObjects(args);
 				CommandObject<Object> commandObject = new CommandObject<>(arguments, BuilderFactory.RAW_OBJECT);
 
 				if (isPipelined()) {
 					pipeline(newJedisResult(getRequiredPipeline().executeCommand(commandObject)));
-				} else {
+				} else if (isQueueing()) {
 					transaction(newJedisResult(getRequiredTransaction().executeCommand(commandObject)));
+				} else {
+					return getRequiredTransaction().executeCommand(commandObject).get();
 				}
 				return null;
 			}
@@ -373,6 +384,10 @@ public boolean isQueueing() {
 		return this.delegate.isQueueing();
 	}
 
+	boolean isWatchOnly() {
+		return this.delegate.isWatchOnly();
+	}
+
 	@Override
 	public boolean isPipelined() {
 		return this.delegate.isPipelined();
@@ -574,7 +589,7 @@ public void pSubscribe(@NonNull MessageListener listener, byte @NonNull [] @NonN
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
 		}
 
-		if (isQueueing() || isPipelined()) {
+		if (isQueueing() || isPipelined() || isWatchOnly()) {
 			throw new InvalidDataAccessApiUsageException("Cannot subscribe in pipeline / transaction mode");
 		}
 
@@ -595,7 +610,7 @@ public void subscribe(@NonNull MessageListener listener, byte @NonNull [] @NonNu
 					"Connection already subscribed; use the connection Subscription to cancel or add new channels");
 		}
 
-		if (isQueueing() || isPipelined()) {
+		if (isQueueing() || isPipelined() || isWatchOnly()) {
 			throw new InvalidDataAccessApiUsageException("Cannot subscribe in pipeline / transaction mode");
 		}
 
@@ -791,6 +806,13 @@ public boolean isQueueing() {
 			return getTransaction() != null;
 		}
 
+		/**
+		 * Indicates whether the connection watches keys without an active {@code MULTI} transaction.
+		 */
+		public boolean isWatchOnly() {
+			return false;
+		}
+
 		/**
 		 * Indicates whether the connection is currently pipelined or not.
 		 */
@@ -977,9 +999,11 @@ public void setClientName(byte @NonNull [] name) {
 		@Override
 		public void watch(byte @NonNull [] @NonNull... keys) {
 
-			if (isMultiExecuted()) {
+			if (isPipelined()) {
+				throw new InvalidDataAccessApiUsageException("WATCH is not supported when pipelining is active");
+			} else if (isMultiExecuted()) {
 				throw new InvalidDataAccessApiUsageException("WATCH is not supported when a transaction is active");
-			} else if (!isQueueing()) {
+			} else if (getTransaction() == null) {
 				setTransaction(getJedis().transaction(false));
 			}
 
@@ -1024,7 +1048,7 @@ public void multi() {
 			}
 
 			if (!isMultiExecuted()) {
-				if (isQueueing()) {
+				if (getTransaction() != null) {
 					// watch was called previously and a transaction is already in progress
 					this.getRequiredTransaction().multi();
 					this.isMultiExecuted = true;
@@ -1074,7 +1098,7 @@ public void discard() {
 		@Override
 		public void openPipeline() {
 
-			if (isQueueing()) {
+			if (isQueueing() || isWatchOnly()) {
 				throw new InvalidDataAccessApiUsageException("Cannot use Pipelining while a transaction is active");
 			}
 
@@ -1120,8 +1144,18 @@ public void close() {
 			this.isMultiExecuted = false;
 		}
 
+		@Override
+		public boolean isQueueing() {
+			return this.isMultiExecuted;
+		}
+
+		@Override
+		public boolean isWatchOnly() {
+			return getTransaction() != null && !this.isMultiExecuted;
+		}
+
 		private boolean isMultiExecuted() {
-			return isQueueing() && this.isMultiExecuted;
+			return this.isMultiExecuted;
 		}
 
 	}

src/main/java/org/springframework/data/redis/connection/jedis/JedisHashCommands.java

@@ -51,6 +51,7 @@
  * @author Mark Paluch
  * @author John Blum
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  * @since 2.0
  */
 @NullUnmarked
@@ -250,7 +251,7 @@ public void hMSet(byte @NonNull [] key, @NonNull Map<byte @NonNull [], byte @Non
 			@Override
 			protected ScanIteration<Entry<byte[], byte[]>> doScan(byte[] key, CursorId cursorId, ScanOptions options) {
 
-				if (connection.isQueueing() || connection.isPipelined()) {
+				if (connection.isQueueing() || connection.isPipelined() || connection.isWatchOnly()) {
 					throw new InvalidDataAccessApiUsageException("'HSCAN' cannot be called in pipeline / transaction mode");
 				}
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisKeyCommands.java

@@ -62,6 +62,7 @@
  * @author ihaohong
  * @author Yordan Tsintsov
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  * @since 2.0
  */
 @NullUnmarked
@@ -186,7 +187,7 @@ public Long touch(byte @NonNull [] @NonNull... keys) {
 			@Override
 			protected ScanIteration<byte[]> doScan(CursorId cursorId, ScanOptions options) {
 
-				if (isQueueing() || isPipelined()) {
+				if (isQueueing() || isPipelined() || connection.isWatchOnly()) {
 					throw new InvalidDataAccessApiUsageException("'SCAN' cannot be called in pipeline / transaction mode");
 				}
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisSetCommands.java

@@ -42,6 +42,7 @@
  * @author Mark Paluch
  * @author Mingi Lee
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  * @since 2.0
  */
 @NullUnmarked
@@ -251,7 +252,7 @@ public Cursor<byte[]> sScan(byte @NonNull [] key, @NonNull ScanOptions options)
 			protected ScanIteration<byte[]> doScan(byte @NonNull [] key, @NonNull CursorId cursorId,
 					@NonNull ScanOptions options) {
 
-				if (connection.isQueueing() || connection.isPipelined()) {
+				if (connection.isQueueing() || connection.isPipelined() || connection.isWatchOnly()) {
 					throw new InvalidDataAccessApiUsageException("'SSCAN' cannot be called in pipeline / transaction mode");
 				}
 

src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java

@@ -55,6 +55,7 @@
  * @author Shyngys Sapraliyev
  * @author John Blum
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  * @since 2.0
  */
 @NullUnmarked
@@ -595,7 +596,7 @@ public Long zUnionStore(byte @NonNull [] destKey, byte @NonNull [] @NonNull... s
 			protected ScanIteration<Tuple> doScan(byte @NonNull [] key, @NonNull CursorId cursorId,
 					@NonNull ScanOptions options) {
 
-				if (connection.isQueueing() || connection.isPipelined()) {
+				if (connection.isQueueing() || connection.isPipelined() || connection.isWatchOnly()) {
 					throw new InvalidDataAccessApiUsageException("'ZSCAN' cannot be called in pipeline / transaction mode");
 				}
 

src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionUnitTests.java

@@ -18,9 +18,11 @@
 import static org.assertj.core.api.Assertions.*;
 import static org.mockito.Mockito.*;
 
+import redis.clients.jedis.AbstractTransaction;
 import redis.clients.jedis.CommandObject;
 import redis.clients.jedis.Connection;
 import redis.clients.jedis.Jedis;
+import redis.clients.jedis.UnifiedJedis;
 
 import java.io.IOException;
 
@@ -32,6 +34,7 @@
 import org.springframework.dao.InvalidDataAccessApiUsageException;
 import org.springframework.dao.InvalidDataAccessResourceUsageException;
 import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase;
+import org.springframework.data.redis.connection.MessageListener;
 import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
 
 /**
@@ -43,9 +46,41 @@
  *
  * @author Christoph Strobl
  * @author Tihomir Mateev
+ * @author Tiefang Hu
  */
 class JedisConnectionUnitTests {
 
+	@Test // GH-3392
+	void subscribeShouldBeRejectedBetweenWatchAndMulti() {
+
+		JedisConnection connection = watchOnlyConnection();
+		MessageListener listener = (message, pattern) -> {};
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> connection.subscribe(listener, "channel".getBytes()));
+	}
+
+	@Test // GH-3392
+	void pSubscribeShouldBeRejectedBetweenWatchAndMulti() {
+
+		JedisConnection connection = watchOnlyConnection();
+		MessageListener listener = (message, pattern) -> {};
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> connection.pSubscribe(listener, "channel-*".getBytes()));
+	}
+
+	private static JedisConnection watchOnlyConnection() {
+
+		UnifiedJedis jedis = mock(UnifiedJedis.class);
+		AbstractTransaction transaction = mock(AbstractTransaction.class);
+		when(jedis.transaction(false)).thenReturn(transaction);
+
+		JedisConnection connection = new JedisConnection(jedis);
+		connection.watch("watched-key".getBytes());
+		return connection;
+	}
+
 	@Nested
 	public class BasicUnitTests extends AbstractConnectionUnitTestBase<Connection> {
 

src/test/java/org/springframework/data/redis/connection/jedis/UnifiedJedisConnectionIntegrationTests.java

@@ -17,12 +17,16 @@
 
 import static org.assertj.core.api.Assertions.*;
 
+import java.util.List;
+
 import redis.clients.jedis.UnifiedJedis;
 
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.junit.jupiter.api.extension.ExtendWith;
 
+import org.springframework.dao.InvalidDataAccessApiUsageException;
+import org.springframework.data.redis.core.ScanOptions;
 import org.springframework.test.context.ContextConfiguration;
 import org.springframework.test.context.junit.jupiter.SpringExtension;
 
@@ -31,6 +35,7 @@
  *
  * @author Tihomir Mateev
  * @author Mark Paluch
+ * @author Tiefang Hu
  */
 @ExtendWith(SpringExtension.class)
 @ContextConfiguration(inheritLocations = false)
@@ -61,4 +66,92 @@ void testNativeConnectionIsJedis() {
 		assertThat(byteConnection.getNativeConnection()).isInstanceOf(UnifiedJedis.class);
 	}
 
+	@Test // GH-3392
+	void readBetweenWatchAndMultiShouldNotBeIncludedInExecResults() {
+
+		connection.set("watched-key", "initial");
+		connection.watch("watched-key".getBytes());
+		assertThat(connection.isQueueing()).isFalse();
+
+		String watchedValue = connection.get("watched-key");
+
+		connection.multi();
+		assertThat(connection.isQueueing()).isTrue();
+		connection.set("watched-key", "updated");
+		connection.set("other-key", "value");
+
+		List<Object> results = connection.exec();
+
+		assertThat(watchedValue).isEqualTo("initial");
+		assertThat(results).containsExactly(true, true);
+		assertThat(connection.isQueueing()).isFalse();
+	}
+
+	@Test // GH-3392
+	void rawCommandBetweenWatchAndMultiShouldUseTheWatchedConnection() {
+
+		byte[] key = "watched-key".getBytes();
+		connection.set("watched-key", "initial");
+		connection.watch(key);
+
+		Object watchedValue = byteConnection.execute("GET", key);
+
+		connection.multi();
+		connection.set("watched-key", "updated");
+
+		List<Object> results = connection.exec();
+
+		assertThat((byte[]) watchedValue).isEqualTo("initial".getBytes());
+		assertThat(results).containsExactly(true);
+	}
+
+	@Test // GH-3392
+	void watchShouldBeRejectedWhilePipelined() {
+
+		byteConnection.openPipeline();
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> connection.watch("watched-key".getBytes()));
+	}
+
+	@Test // GH-3392
+	void scanShouldBeRejectedBetweenWatchAndMulti() {
+
+		byte[] key = "watched-key".getBytes();
+		connection.watch(key);
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> byteConnection.scan(ScanOptions.NONE));
+	}
+
+	@Test // GH-3392
+	void hScanShouldBeRejectedBetweenWatchAndMulti() {
+
+		byte[] key = "watched-key".getBytes();
+		connection.watch(key);
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> byteConnection.hScan(key, ScanOptions.NONE));
+	}
+
+	@Test // GH-3392
+	void sScanShouldBeRejectedBetweenWatchAndMulti() {
+
+		byte[] key = "watched-key".getBytes();
+		connection.watch(key);
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> byteConnection.sScan(key, ScanOptions.NONE));
+	}
+
+	@Test // GH-3392
+	void zScanShouldBeRejectedBetweenWatchAndMulti() {
+
+		byte[] key = "watched-key".getBytes();
+		connection.watch(key);
+
+		assertThatExceptionOfType(InvalidDataAccessApiUsageException.class)
+				.isThrownBy(() -> byteConnection.zScan(key, ScanOptions.NONE));
+	}
+
 }