diff --git a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java index fa90e203f..016b1f575 100644 --- a/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java @@ -229,7 +229,6 @@ public interface RedisStreamCommands { /** * @author Christoph Strobl - * @author Dengliming * @since 2.3 */ class XClaimOptions { @@ -240,10 +239,9 @@ public interface RedisStreamCommands { private final @Nullable Instant unixTime; private final @Nullable Long retryCount; private final boolean force; - private final boolean justId; private XClaimOptions(List ids, Duration minIdleTime, @Nullable Duration idleTime, - @Nullable Instant unixTime, @Nullable Long retryCount, boolean force, boolean justId) { + @Nullable Instant unixTime, @Nullable Long retryCount, boolean force) { this.ids = new ArrayList<>(ids); this.minIdleTime = minIdleTime; @@ -251,7 +249,6 @@ public interface RedisStreamCommands { this.unixTime = unixTime; this.retryCount = retryCount; this.force = force; - this.justId = justId; } /** @@ -284,7 +281,7 @@ public interface RedisStreamCommands { * @return {@code this}. */ public XClaimOptions idle(Duration idleTime) { - return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId); + return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force); } /** @@ -295,7 +292,7 @@ public interface RedisStreamCommands { * @return {@code this}. */ public XClaimOptions time(Instant unixTime) { - return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId); + return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force); } /** @@ -305,7 +302,7 @@ public interface RedisStreamCommands { * @return new instance of {@link XClaimOptions}. */ public XClaimOptions retryCount(long retryCount) { - return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, justId); + return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force); } /** @@ -315,16 +312,7 @@ public interface RedisStreamCommands { * @return new instance of {@link XClaimOptions}. */ public XClaimOptions force() { - return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, true, justId); - } - - /** - * Set the JUSTID flag. - * - * @return new instance of {@link XClaimOptions}. - */ - public XClaimOptions justId() { - return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, force, true); + return new XClaimOptions(ids, minIdleTime, idleTime, unixTime, retryCount, true); } /** @@ -393,15 +381,6 @@ public interface RedisStreamCommands { return force; } - /** - * Get the JUSTID flag. - * - * @return - */ - public boolean isJustId() { - return justId; - } - public static class XClaimOptionsBuilder { private final Duration minIdleTime; @@ -425,7 +404,7 @@ public interface RedisStreamCommands { .map(it -> it instanceof RecordId ? (RecordId) it : RecordId.of(it.toString())) .collect(Collectors.toList()); - return new XClaimOptions(idList, minIdleTime, null, null, null, false, false); + return new XClaimOptions(idList, minIdleTime, null, null, null, false); } /** diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java index 69d531afb..8d01fbc4e 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java @@ -128,8 +128,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands { String[] ids = command.getOptions().getIdsAsStringArray(); io.lettuce.core.Consumer from = io.lettuce.core.Consumer .from(ByteUtils.getByteBuffer(command.getGroupName()), ByteUtils.getByteBuffer(command.getNewOwner())); - XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions()); - + XClaimArgs args = StreamConverters.toXClaimArgs(command.getOptions()).justid(); Flux result = cmd.xclaim(command.getKey(), from, args, ids).map(it -> RecordId.of(it.getId())); return new CommandResponse<>(command, result); })); diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java index 6066b0fe4..55dd4c9b8 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java @@ -135,7 +135,7 @@ class LettuceStreamCommands implements RedisStreamCommands { String[] ids = options.getIdsAsStringArray(); io.lettuce.core.Consumer from = io.lettuce.core.Consumer.from(LettuceConverters.toBytes(group), LettuceConverters.toBytes(newOwner)); - XClaimArgs args = StreamConverters.toXClaimArgs(options); + XClaimArgs args = StreamConverters.toXClaimArgs(options).justid(); try { if (isPipelined()) { diff --git a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java index 6152bd64e..c7e765f76 100644 --- a/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java +++ b/src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java @@ -236,9 +236,6 @@ class StreamConverters { if (source.getUnixTime() != null) { args.time(source.getUnixTime()); } - if (source.isJustId()) { - args.justid(); - } return args; } diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java index 38f9ea657..0f3c7f59d 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java @@ -20,27 +20,30 @@ import static org.mockito.Mockito.*; import io.lettuce.core.RedisClient; import io.lettuce.core.XAddArgs; +import io.lettuce.core.XClaimArgs; import io.lettuce.core.api.StatefulRedisConnection; import io.lettuce.core.api.async.RedisAsyncCommands; import io.lettuce.core.api.sync.RedisCommands; import io.lettuce.core.codec.RedisCodec; import java.lang.reflect.InvocationTargetException; +import java.time.Duration; import java.util.Collections; import org.junit.Before; import org.junit.Test; import org.junit.runner.RunWith; import org.junit.runners.Suite; - import org.mockito.ArgumentCaptor; import org.springframework.dao.InvalidDataAccessResourceUsageException; import org.springframework.data.redis.connection.AbstractConnectionUnitTestBase; import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption; import org.springframework.data.redis.connection.RedisStreamCommands.XAddOptions; +import org.springframework.data.redis.connection.RedisStreamCommands.XClaimOptions; import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettuceConnectionUnitTests; import org.springframework.data.redis.connection.lettuce.LettuceConnectionUnitTestSuite.LettucePipelineConnectionUnitTests; import org.springframework.data.redis.connection.stream.MapRecord; +import org.springframework.test.util.ReflectionTestUtils; /** * @author Christoph Strobl @@ -173,15 +176,51 @@ public class LettuceConnectionUnitTestSuite { @Test // DATAREDIS-1122 public void xaddShouldHonorMaxlen() { - MapRecord record = MapRecord.create("key".getBytes(), - Collections.emptyMap()); + MapRecord record = MapRecord.create("key".getBytes(), Collections.emptyMap()); connection.streamCommands().xAdd(record, XAddOptions.maxlen(100)); ArgumentCaptor args = ArgumentCaptor.forClass(XAddArgs.class); - verify(syncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap()); + if (connection.isPipelined()) { + verify(asyncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap()); + } else { + verify(syncCommandsMock, times(1)).xadd(any(), args.capture(), anyMap()); + } assertThat(args.getValue()).extracting("maxlen").isEqualTo(100L); } + + @Test // DATAREDIS-1226 + public void xClaimShouldNotAddJustIdFlagToArgs() { + + connection.streamCommands().xClaim("key".getBytes(), "group", "owner", + XClaimOptions.minIdle(Duration.ofMillis(100)).ids("1-1")); + ArgumentCaptor args = ArgumentCaptor.forClass(XClaimArgs.class); + + if (connection.isPipelined()) { + verify(asyncCommandsMock).xclaim(any(), any(), args.capture(), any()); + } else { + verify(syncCommandsMock).xclaim(any(), any(), args.capture(), any()); + } + + assertThat(ReflectionTestUtils.getField(args.getValue(), "justid")).isEqualTo(false); + + } + + @Test // DATAREDIS-1226 + public void xClaimJustIdShouldAddJustIdFlagToArgs() { + + connection.streamCommands().xClaimJustId("key".getBytes(), "group", "owner", + XClaimOptions.minIdle(Duration.ofMillis(100)).ids("1-1")); + ArgumentCaptor args = ArgumentCaptor.forClass(XClaimArgs.class); + + if (connection.isPipelined()) { + verify(asyncCommandsMock).xclaim(any(), any(), args.capture(), any()); + } else { + verify(syncCommandsMock).xclaim(any(), any(), args.capture(), any()); + } + + assertThat(ReflectionTestUtils.getField(args.getValue(), "justid")).isEqualTo(true); + } } public static class LettucePipelineConnectionUnitTests extends LettuceConnectionUnitTests { diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java index a524f83f2..7a3f14b13 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java @@ -502,7 +502,7 @@ public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsT StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) // .delayElements(Duration.ofMillis(5)).next() // .flatMapMany(record -> connection.streamCommands().xClaimJustId(KEY_1_BBUFFER, "my-group", "my-consumer", - XClaimOptions.minIdle(Duration.ofMillis(1)).ids(record.getId()).justId()) + XClaimOptions.minIdle(Duration.ofMillis(1)).ids(record.getId())) ).as(StepVerifier::create) // .assertNext(it -> assertThat(it.getValue()).isEqualTo(expected)) // .verifyComplete();