diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java index 1a6e505f46..3406f9af39 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisConnection.java +++ b/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 response = pipelineFunction.apply(JedisInvoker.createCommands(getRequiredTransaction())); + Object result = response.get(); + + return result != null ? converter.convert(result) : nullDefault.get(); + } + if (isQueueing()) { Response 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 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; } } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisHashCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisHashCommands.java index 987d2c42c7..ea2f0d7dbb 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisHashCommands.java +++ b/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> 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"); } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisKeyCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisKeyCommands.java index a7368a2af8..c7f6ee8285 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisKeyCommands.java +++ b/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 doScan(CursorId cursorId, ScanOptions options) { - if (isQueueing() || isPipelined()) { + if (isQueueing() || isPipelined() || connection.isWatchOnly()) { throw new InvalidDataAccessApiUsageException("'SCAN' cannot be called in pipeline / transaction mode"); } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisSetCommands.java index 4ee7c34c2b..8389650201 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisSetCommands.java +++ b/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 sScan(byte @NonNull [] key, @NonNull ScanOptions options) protected ScanIteration 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"); } diff --git a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java b/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java index e3be9bb7a1..2a689a74e6 100644 --- a/src/main/java/org/springframework/data/redis/connection/jedis/JedisZSetCommands.java +++ b/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 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"); } diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionUnitTests.java index 9654345f49..b93b5bdba2 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/JedisConnectionUnitTests.java +++ b/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 { diff --git a/src/test/java/org/springframework/data/redis/connection/jedis/UnifiedJedisConnectionIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/jedis/UnifiedJedisConnectionIntegrationTests.java index 0c42effc38..e9ba2b22c6 100644 --- a/src/test/java/org/springframework/data/redis/connection/jedis/UnifiedJedisConnectionIntegrationTests.java +++ b/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 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 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)); + } + }