Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

DATAREDIS-1226 - Polishing.

Browse Source 
Remove justId from XClaimOptions and set the flag inside xClaimJustId to avoid errors when processing justId inside xClaim having a different return type.

Original Pull Request: #567
This commit is contained in:
Christoph Strobl
2020-10-07 10:44:38 +02:00
parent d2dce680cc
commit 3920890c01
6 changed files with 52 additions and 38 deletions
Download Patch File Download Diff File Expand all files Collapse all files

33
src/main/java/org/springframework/data/redis/connection/RedisStreamCommands.java
View File

@@ -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<RecordId> 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);
}
/**

3
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommands.java
View File

@@ -128,8 +128,7 @@ class LettuceReactiveStreamCommands implements ReactiveStreamCommands {
String[] ids = command.getOptions().getIdsAsStringArray();
io.lettuce.core.Consumer<ByteBuffer> 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<RecordId> result = cmd.xclaim(command.getKey(), from, args, ids).map(it -> RecordId.of(it.getId()));
return new CommandResponse<>(command, result);
}));

2
src/main/java/org/springframework/data/redis/connection/lettuce/LettuceStreamCommands.java
View File

@@ -135,7 +135,7 @@ class LettuceStreamCommands implements RedisStreamCommands {
String[] ids = options.getIdsAsStringArray();
io.lettuce.core.Consumer<byte[]> 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()) {

3
src/main/java/org/springframework/data/redis/connection/lettuce/StreamConverters.java
View File

@@ -236,9 +236,6 @@ class StreamConverters {
if (source.getUnixTime() != null) {
args.time(source.getUnixTime());
}
if (source.isJustId()) {
args.justid();
}
return args;
}

47
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionUnitTestSuite.java
View File

@@ -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<byte[], byte[], byte[]> record = MapRecord.create("key".getBytes(),
Collections.emptyMap());
MapRecord<byte[], byte[], byte[]> record = MapRecord.create("key".getBytes(), Collections.emptyMap());
connection.streamCommands().xAdd(record, XAddOptions.maxlen(100));
ArgumentCaptor<XAddArgs> 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<XClaimArgs> 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<XClaimArgs> 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 {

2
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveStreamCommandsTests.java
View File

@@ -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();