DATAREDIS-864 - Add support for Redis Streams.
We now support Redis Streams to add, read and consume stream records. We introduced StreamOperations, BoundStreamOperations, and ReactiveStreamOperations to interact with Redis Streams using imperative and reactive programming models. Record represents items within a stream. There are various flavors of Stream Records:
* MapRecord (maps to the hash body used in stream messages).
* Binary MapRecord: byte[] and ByteBuffer variants of MapRecord.
* ObjectRecord: Simple and Complex Objects mapped onto the stream body hash using ObjectHashMapper.
Redis Streams are supported for the Lettuce client only as Jedis has not received yet Redis Stream support.
Messages can be created as Map or as object:
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value"));
redisTemplate.opsForStream().add(ObjectRecord.create("my-logins", new LoginEvent(…)));
Streams can be consumed by using a StreamMessageListenerContainer that allows for stream subscriptions or a StreamReceiver.
Synchronous Message Listener:
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
.create(connectionFactory);
container.start();
Subscription subscription = container.receive(StreamOffset.fromStart("my-stream"), record -> … );
Reactive Message Receiver:
StreamReceiverOptions<String, ObjectRecord<String, LoginEvent>> receiverOptions = StreamReceiverOptions.builder()
.targetType(LoginEvent.class).build();
StreamReceiver<String, ObjectRecord<String, LoginEvent>> receiver = StreamReceiver.create(connectionFactory, receiverOptions);
Flux<ObjectRecord<String, LoginEvent>> messages = receiver.receive(StreamOffset.fromStart("my-logins"));
Original Pull Request: #356
This commit is contained in:
Mark Paluch
committed by
Christoph Strobl
Christoph Strobl
parent
f2544a253a
commit
65754623f6
@@ -40,6 +40,7 @@ public class RedisTestProfileValueSource implements ProfileValueSource {
|
||||
private static final String REDIS_28 = "2.8";
|
||||
private static final String REDIS_30 = "3.0";
|
||||
private static final String REDIS_32 = "3.2";
|
||||
private static final String REDIS_50 = "4.9";
|
||||
private static final String REDIS_VERSION_KEY = "redisVersion";
|
||||
|
||||
private static RedisTestProfileValueSource INSTANCE;
|
||||
@@ -96,6 +97,10 @@ public class RedisTestProfileValueSource implements ProfileValueSource {
|
||||
return System.getProperty(key);
|
||||
}
|
||||
|
||||
if (redisVersion.compareTo(RedisVersionUtils.parseVersion(REDIS_50)) >= 0) {
|
||||
return REDIS_50;
|
||||
}
|
||||
|
||||
if (redisVersion.compareTo(RedisVersionUtils.parseVersion(REDIS_32)) >= 0) {
|
||||
return REDIS_32;
|
||||
}
|
||||
|
||||
@@ -64,6 +64,10 @@ import org.springframework.data.redis.RedisVersionUtils;
|
||||
import org.springframework.data.redis.TestCondition;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
|
||||
import org.springframework.data.redis.connection.RedisListCommands.Position;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
|
||||
import org.springframework.data.redis.connection.RedisStringCommands.SetOption;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
|
||||
@@ -1156,11 +1160,11 @@ public abstract class AbstractConnectionIntegrationTests {
|
||||
@IfProfileValue(name = "redisVersion", value = "4.0+")
|
||||
public void unlinkReturnsNrOfKeysRemoved() {
|
||||
|
||||
connection.set("unlink.this", "Can't track this!");
|
||||
actual.add(connection.set("unlink.this", "Can't track this!"));
|
||||
|
||||
actual.add(connection.unlink("unlink.this", "unlink.that"));
|
||||
|
||||
verifyResults(Arrays.asList(new Object[] { 1L }));
|
||||
verifyResults(Arrays.asList(new Object[] { true, 1L }));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-693
|
||||
@@ -2999,6 +3003,101 @@ public abstract class AbstractConnectionIntegrationTests {
|
||||
assertThat((List<Long>) results.get(1), contains(100L, -56L));
|
||||
}
|
||||
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
@IfProfileValue(name = "redisVersion", value = "5.0")
|
||||
@WithRedisDriver({ RedisDriver.LETTUCE })
|
||||
public void xAddShouldCreateStream() {
|
||||
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
actual.add(connection.type(KEY_1));
|
||||
|
||||
List<Object> results = getResults();
|
||||
assertThat(results, hasSize(2));
|
||||
assertThat((String) results.get(0), containsString("-"));
|
||||
assertThat(results.get(1), is(DataType.STREAM));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
@IfProfileValue(name = "redisVersion", value = "5.0")
|
||||
@WithRedisDriver({ RedisDriver.LETTUCE })
|
||||
public void xReadShouldReadMessage() {
|
||||
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
actual.add(connection.xReadAsString(StreamOffset.create(KEY_1, ReadOffset.from("0"))));
|
||||
|
||||
List<Object> results = getResults();
|
||||
|
||||
List<StreamMessage<String, String>> messages = (List) results.get(1);
|
||||
|
||||
assertThat(messages.get(0).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
@IfProfileValue(name = "redisVersion", value = "5.0")
|
||||
@WithRedisDriver({ RedisDriver.LETTUCE })
|
||||
public void xReadGroupShouldReadMessage() {
|
||||
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
actual.add(connection.xGroupCreate(KEY_1, ReadOffset.from("0"), "my-group"));
|
||||
actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
|
||||
actual.add(connection.xReadGroupAsString(Consumer.from("my-group", "my-consumer"), StreamOffset.create(KEY_1, ReadOffset.lastConsumed())));
|
||||
|
||||
List<Object> results = getResults();
|
||||
|
||||
List<StreamMessage<String, String>> messages = (List) results.get(2);
|
||||
|
||||
assertThat(messages.get(0).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
|
||||
assertThat((List<StreamMessage>) results.get(3), is(empty()));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
@IfProfileValue(name = "redisVersion", value = "5.0")
|
||||
@WithRedisDriver({ RedisDriver.LETTUCE })
|
||||
public void xRangeShouldReportMessages() {
|
||||
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3)));
|
||||
actual.add(connection.xRange(KEY_1, org.springframework.data.domain.Range.unbounded()));
|
||||
|
||||
List<Object> results = getResults();
|
||||
assertThat(results, hasSize(3));
|
||||
assertThat((String) results.get(0), containsString("-"));
|
||||
|
||||
List<StreamMessage<String, String>> messages = (List) results.get(2);
|
||||
|
||||
assertThat(messages.get(0).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
|
||||
assertThat(messages.get(1).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(1).getBody(), is(Collections.singletonMap(KEY_3, VALUE_3)));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
@IfProfileValue(name = "redisVersion", value = "5.0")
|
||||
@WithRedisDriver({ RedisDriver.LETTUCE })
|
||||
public void xRevRangeShouldReportMessages() {
|
||||
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
actual.add(connection.xAdd(KEY_1, Collections.singletonMap(KEY_3, VALUE_3)));
|
||||
actual.add(connection.xRevRange(KEY_1, org.springframework.data.domain.Range.unbounded()));
|
||||
|
||||
List<Object> results = getResults();
|
||||
assertThat(results, hasSize(3));
|
||||
assertThat((String) results.get(0), containsString("-"));
|
||||
|
||||
List<StreamMessage<String, String>> messages = (List) results.get(2);
|
||||
|
||||
assertThat(messages.get(0).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(0).getBody(), is(Collections.singletonMap(KEY_3, VALUE_3)));
|
||||
|
||||
assertThat(messages.get(1).getStream(), is(KEY_1));
|
||||
assertThat(messages.get(1).getBody(), is(Collections.singletonMap(KEY_2, VALUE_2)));
|
||||
}
|
||||
|
||||
protected void verifyResults(List<Object> expected) {
|
||||
assertEquals(expected, getResults());
|
||||
}
|
||||
|
||||
@@ -29,6 +29,7 @@ import org.junit.ClassRule;
|
||||
import org.junit.Ignore;
|
||||
import org.junit.Rule;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.springframework.dao.InvalidDataAccessApiUsageException;
|
||||
import org.springframework.data.redis.connection.AbstractConnectionIntegrationTests;
|
||||
import org.springframework.data.redis.connection.RedisSentinelConfiguration;
|
||||
@@ -37,7 +38,9 @@ import org.springframework.data.redis.connection.RedisServer;
|
||||
import org.springframework.data.redis.connection.ReturnType;
|
||||
import org.springframework.data.redis.test.util.MinimumRedisVersionRule;
|
||||
import org.springframework.data.redis.test.util.RedisSentinelRule;
|
||||
import org.springframework.data.redis.test.util.RelaxedJUnit4ClassRunner;
|
||||
import org.springframework.test.annotation.IfProfileValue;
|
||||
import org.springframework.test.context.ContextConfiguration;
|
||||
import org.springframework.test.util.ReflectionTestUtils;
|
||||
|
||||
/**
|
||||
@@ -45,6 +48,8 @@ import org.springframework.test.util.ReflectionTestUtils;
|
||||
* @author Thomas Darimont
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
@RunWith(RelaxedJUnit4ClassRunner.class)
|
||||
@ContextConfiguration("JedisConnectionIntegrationTests-context.xml")
|
||||
public class JedisSentinelIntegrationTests extends AbstractConnectionIntegrationTests {
|
||||
|
||||
private static final String MASTER_NAME = "mymaster";
|
||||
|
||||
@@ -83,7 +83,7 @@ public abstract class LettuceReactiveCommandsTestsBase {
|
||||
RedisClusterCommands<String, String> nativeCommands;
|
||||
RedisClusterCommands<ByteBuffer, ByteBuffer> nativeBinaryCommands;
|
||||
|
||||
@Parameterized.Parameters(name = "{2}")
|
||||
@Parameterized.Parameters(name = "{3}")
|
||||
public static List<Object[]> parameters() {
|
||||
|
||||
LettuceRedisClientProvider standalone = LettuceRedisClientProvider.local();
|
||||
|
||||
@@ -0,0 +1,173 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.connection.lettuce;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import io.lettuce.core.XReadArgs;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import java.util.Collections;
|
||||
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.domain.Range;
|
||||
import org.springframework.data.redis.RedisTestProfileValueSource;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
|
||||
/**
|
||||
* Integration tests for {@link LettuceReactiveStreamCommands}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class LettuceReactiveStreamCommandsTests extends LettuceReactiveCommandsTestsBase {
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
|
||||
// TODO: Upgrade to 5.0
|
||||
assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9"));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xAddShouldAddMessage() {
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_2_BBUFFER, VALUE_2_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xLen(KEY_1_BBUFFER) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1L) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xDelShouldRemoveMessage() {
|
||||
|
||||
String messageId = connection.streamCommands()
|
||||
.xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_2_BBUFFER, VALUE_2_BBUFFER)).block();
|
||||
|
||||
connection.streamCommands().xDel(KEY_1_BBUFFER, messageId) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1L) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xLen(KEY_1_BBUFFER) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(0L) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xRangeShouldReportMessages() {
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_1_BBUFFER, VALUE_1_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_2_BBUFFER, VALUE_2_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xRange(KEY_1_BBUFFER, Range.unbounded()) //
|
||||
.as(StepVerifier::create) //
|
||||
.assertNext(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
|
||||
assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
|
||||
|
||||
}) //
|
||||
.expectNextCount(1).verifyComplete();
|
||||
|
||||
connection.streamCommands().xRange(KEY_1_BBUFFER, Range.unbounded(), Limit.limit().count(1)) //
|
||||
.as(StepVerifier::create) //
|
||||
.assertNext(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
|
||||
assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xReadShouldReadMessage() {
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_1_BBUFFER, VALUE_1_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xRead(StreamOffset.create(KEY_1_BBUFFER, ReadOffset.from("0-0"))) //
|
||||
.as(StepVerifier::create) //
|
||||
.assertNext(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
|
||||
assertThat(it.getBody()).containsEntry(KEY_1_BBUFFER, VALUE_1_BBUFFER);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xReadGroupShouldReadMessage() {
|
||||
|
||||
String initialMessage = nativeCommands.xadd(KEY_1, KEY_1, VALUE_1);
|
||||
nativeCommands.xgroupCreate(XReadArgs.StreamOffset.from(KEY_1, initialMessage), "my-group");
|
||||
|
||||
nativeCommands.xadd(KEY_1, KEY_2, VALUE_2);
|
||||
|
||||
connection.streamCommands()
|
||||
.xReadGroup(Consumer.from("my-group", "my-consumer"),
|
||||
StreamOffset.create(KEY_1_BBUFFER, ReadOffset.lastConsumed())) //
|
||||
.as(StepVerifier::create) //
|
||||
.assertNext(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
|
||||
assertThat(it.getBody()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void xRevRangeShouldReportMessages() {
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_1_BBUFFER, VALUE_1_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xAdd(KEY_1_BBUFFER, Collections.singletonMap(KEY_2_BBUFFER, VALUE_2_BBUFFER)) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.verifyComplete();
|
||||
|
||||
connection.streamCommands().xRevRange(KEY_1_BBUFFER, Range.unbounded(), Limit.limit().count(1)) //
|
||||
.as(StepVerifier::create) //
|
||||
.assertNext(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo(KEY_1_BBUFFER);
|
||||
assertThat(it.getBody()).containsEntry(KEY_2_BBUFFER, VALUE_2_BBUFFER);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
}
|
||||
@@ -27,7 +27,7 @@ import org.springframework.data.redis.RawObjectFactory;
|
||||
import org.springframework.data.redis.SettingsUtils;
|
||||
import org.springframework.data.redis.StringObjectFactory;
|
||||
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
|
||||
import org.springframework.data.redis.connection.jedis.JedisConnectionFactory;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.GenericToStringSerializer;
|
||||
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
|
||||
@@ -65,7 +65,7 @@ abstract public class AbstractOperationsTestParams {
|
||||
throw new RuntimeException("Cannot init XStream", ex);
|
||||
}
|
||||
|
||||
JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory(standaloneConfiguration);
|
||||
LettuceConnectionFactory jedisConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration);
|
||||
jedisConnectionFactory.afterPropertiesSet();
|
||||
|
||||
RedisTemplate<String, String> stringTemplate = new StringRedisTemplate();
|
||||
|
||||
@@ -0,0 +1,228 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.core;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.Parameterized;
|
||||
import org.junit.runners.Parameterized.Parameters;
|
||||
import org.springframework.data.domain.Range;
|
||||
import org.springframework.data.domain.Range.Bound;
|
||||
import org.springframework.data.redis.ConnectionFactoryTracker;
|
||||
import org.springframework.data.redis.ObjectFactory;
|
||||
import org.springframework.data.redis.PersonObjectFactory;
|
||||
import org.springframework.data.redis.RedisTestProfileValueSource;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
|
||||
/**
|
||||
* Integration tests for {@link DefaultReactiveStreamOperations}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
@RunWith(Parameterized.class)
|
||||
@SuppressWarnings("unchecked")
|
||||
public class DefaultReactiveStreamOperationsTests<K, V> {
|
||||
|
||||
private final ReactiveRedisTemplate<K, V> redisTemplate;
|
||||
private final ReactiveStreamOperations<K, V> streamOperations;
|
||||
|
||||
private final ObjectFactory<K> keyFactory;
|
||||
private final ObjectFactory<V> valueFactory;
|
||||
|
||||
@Parameters(name = "{4}")
|
||||
public static Collection<Object[]> testParams() {
|
||||
return ReactiveOperationsTestParams.testParams();
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void cleanUp() {
|
||||
ConnectionFactoryTracker.cleanUp();
|
||||
}
|
||||
|
||||
/**
|
||||
* @param redisTemplate
|
||||
* @param keyFactory
|
||||
* @param valueFactory
|
||||
* @param label parameterized test label, no further use besides that.
|
||||
*/
|
||||
public DefaultReactiveStreamOperationsTests(ReactiveRedisTemplate<K, V> redisTemplate, ObjectFactory<K> keyFactory,
|
||||
ObjectFactory<V> valueFactory, RedisSerializer serializer, String label) {
|
||||
|
||||
// Currently, only Lettuce supports Redis Streams.
|
||||
// See https://github.com/xetorthio/jedis/issues/1820
|
||||
assumeTrue(redisTemplate.getConnectionFactory() instanceof LettuceConnectionFactory);
|
||||
|
||||
// TODO: Change to 5.0 after Redis 5 GA
|
||||
assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9"));
|
||||
|
||||
this.redisTemplate = redisTemplate;
|
||||
this.streamOperations = redisTemplate.opsForStream();
|
||||
this.keyFactory = keyFactory;
|
||||
this.valueFactory = valueFactory;
|
||||
|
||||
ConnectionFactoryTracker.add(redisTemplate.getConnectionFactory());
|
||||
}
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
|
||||
RedisConnectionFactory connectionFactory = (RedisConnectionFactory) redisTemplate.getConnectionFactory();
|
||||
RedisConnection connection = connectionFactory.getConnection();
|
||||
connection.flushAll();
|
||||
connection.close();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void addShouldAddMessage() {
|
||||
|
||||
assumeFalse(valueFactory instanceof PersonObjectFactory);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
|
||||
streamOperations.range(key, Range.unbounded()) //
|
||||
.as(StepVerifier::create) //
|
||||
.consumeNextWith(actual -> {
|
||||
|
||||
assertThat(actual.getId()).isEqualTo(messageId);
|
||||
assertThat(actual.getStream()).isEqualTo(key);
|
||||
|
||||
if (!(key instanceof byte[] || value instanceof byte[])) {
|
||||
assertThat(actual.getBody()).containsEntry(key, value);
|
||||
}
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void rangeShouldReportMessages() {
|
||||
|
||||
assumeFalse(valueFactory instanceof PersonObjectFactory);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId1 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
String messageId2 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
|
||||
streamOperations
|
||||
.range(key, Range.from(Bound.inclusive(messageId1)).to(Bound.inclusive(messageId2)), Limit.limit().count(1)) //
|
||||
.as(StepVerifier::create) //
|
||||
.consumeNextWith(actual -> {
|
||||
|
||||
assertThat(actual.getId()).isEqualTo(messageId1);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void reverseRangeShouldReportMessages() {
|
||||
|
||||
assumeFalse(valueFactory instanceof PersonObjectFactory);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId1 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
String messageId2 = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
|
||||
streamOperations.reverseRange(key, Range.unbounded()).map(StreamMessage::getId) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(messageId2, messageId1) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void readShouldReadMessage() {
|
||||
|
||||
assumeFalse(valueFactory instanceof PersonObjectFactory);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId = streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
|
||||
streamOperations.read(StreamOffset.create(key, ReadOffset.from("0-0"))) //
|
||||
.as(StepVerifier::create) //
|
||||
.consumeNextWith(actual -> {
|
||||
|
||||
assertThat(actual.getId()).isEqualTo(messageId);
|
||||
assertThat(actual.getStream()).isEqualTo(key);
|
||||
|
||||
if (!(key instanceof byte[] || value instanceof byte[])) {
|
||||
assertThat(actual.getBody()).containsEntry(key, value);
|
||||
}
|
||||
}).verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void readShouldReadMessages() {
|
||||
|
||||
assumeFalse(valueFactory instanceof PersonObjectFactory);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
streamOperations.add(key, Collections.singletonMap(key, value)).block();
|
||||
|
||||
streamOperations.read(StreamReadOptions.empty().count(2), StreamOffset.create(key, ReadOffset.from("0-0"))) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNextCount(2) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void sizeShouldReportStreamSize() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
streamOperations.add(key, Collections.singletonMap(key, value)).as(StepVerifier::create).expectNextCount(1)
|
||||
.verifyComplete();
|
||||
streamOperations.size(key) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(1L) //
|
||||
.verifyComplete();
|
||||
|
||||
streamOperations.add(key, Collections.singletonMap(key, value)).as(StepVerifier::create).expectNextCount(1)
|
||||
.verifyComplete();
|
||||
streamOperations.size(key) //
|
||||
.as(StepVerifier::create) //
|
||||
.expectNext(2L) //
|
||||
.verifyComplete();
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,225 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.core;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import java.util.Collection;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.junit.runner.RunWith;
|
||||
import org.junit.runners.Parameterized;
|
||||
import org.junit.runners.Parameterized.Parameters;
|
||||
import org.springframework.data.domain.Range;
|
||||
import org.springframework.data.domain.Range.Bound;
|
||||
import org.springframework.data.redis.ConnectionFactoryTracker;
|
||||
import org.springframework.data.redis.ObjectFactory;
|
||||
import org.springframework.data.redis.RedisTestProfileValueSource;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
|
||||
/**
|
||||
* Integration test of {@link DefaultStreamOperations}
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
@RunWith(Parameterized.class)
|
||||
public class DefaultStreamOperationsTests<K, V> {
|
||||
|
||||
private RedisTemplate<K, V> redisTemplate;
|
||||
|
||||
private ObjectFactory<K> keyFactory;
|
||||
|
||||
private ObjectFactory<V> valueFactory;
|
||||
|
||||
private StreamOperations<K, V> streamOps;
|
||||
|
||||
public DefaultStreamOperationsTests(RedisTemplate<K, V> redisTemplate, ObjectFactory<K> keyFactory,
|
||||
ObjectFactory<V> valueFactory) {
|
||||
|
||||
// Currently, only Lettuce supports Redis Streams.
|
||||
// See https://github.com/xetorthio/jedis/issues/1820
|
||||
assumeTrue(redisTemplate.getConnectionFactory() instanceof LettuceConnectionFactory);
|
||||
|
||||
// TODO: Change to 5.0 after Redis 5 GA
|
||||
assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "4.9"));
|
||||
|
||||
this.redisTemplate = redisTemplate;
|
||||
this.keyFactory = keyFactory;
|
||||
this.valueFactory = valueFactory;
|
||||
|
||||
ConnectionFactoryTracker.add(redisTemplate.getConnectionFactory());
|
||||
}
|
||||
|
||||
@Parameters
|
||||
public static Collection<Object[]> testParams() {
|
||||
return AbstractOperationsTestParams.testParams();
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void cleanUp() {
|
||||
ConnectionFactoryTracker.cleanUp();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
streamOps = redisTemplate.opsForStream();
|
||||
|
||||
redisTemplate.execute((RedisCallback<Object>) connection -> {
|
||||
connection.flushDb();
|
||||
return null;
|
||||
});
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void addShouldAddMessage() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.range(key, Range.unbounded());
|
||||
|
||||
assertThat(messages).hasSize(1);
|
||||
|
||||
StreamMessage<K, V> message = messages.get(0);
|
||||
|
||||
assertThat(message.getId()).isEqualTo(messageId);
|
||||
assertThat(message.getStream()).isEqualTo(key);
|
||||
|
||||
if (!(key instanceof byte[] || value instanceof byte[])) {
|
||||
assertThat(message.getBody()).containsEntry(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void rangeShouldReportMessages() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId1 = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
String messageId2 = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.range(key,
|
||||
Range.from(Bound.inclusive(messageId1)).to(Bound.inclusive(messageId2)), Limit.limit().count(1));
|
||||
|
||||
assertThat(messages).hasSize(1);
|
||||
|
||||
StreamMessage<K, V> message = messages.get(0);
|
||||
|
||||
assertThat(message.getId()).isEqualTo(messageId1);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void reverseRangeShouldReportMessages() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId1 = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
String messageId2 = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.reverseRange(key, Range.unbounded());
|
||||
|
||||
assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void readShouldReadMessage() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.read(StreamOffset.create(key, ReadOffset.from("0-0")));
|
||||
|
||||
assertThat(messages).hasSize(1);
|
||||
|
||||
StreamMessage<K, V> message = messages.get(0);
|
||||
|
||||
assertThat(message.getId()).isEqualTo(messageId);
|
||||
assertThat(message.getStream()).isEqualTo(key);
|
||||
|
||||
if (!(key instanceof byte[] || value instanceof byte[])) {
|
||||
assertThat(message.getBody()).containsEntry(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void readShouldReadMessages() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
streamOps.add(key, Collections.singletonMap(key, value));
|
||||
streamOps.add(key, Collections.singletonMap(key, value));
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.read(StreamReadOptions.empty().count(2),
|
||||
StreamOffset.create(key, ReadOffset.from("0-0")));
|
||||
|
||||
assertThat(messages).hasSize(2);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void readShouldReadMessageWithConsumerGroup() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
String messageId = streamOps.add(key, Collections.singletonMap(key, value));
|
||||
streamOps.createGroup(key, ReadOffset.from("0-0"), "my-group");
|
||||
|
||||
List<StreamMessage<K, V>> messages = streamOps.read(Consumer.from("my-group", "my-consumer"),
|
||||
StreamOffset.create(key, ReadOffset.lastConsumed()));
|
||||
|
||||
assertThat(messages).hasSize(1);
|
||||
|
||||
StreamMessage<K, V> message = messages.get(0);
|
||||
|
||||
assertThat(message.getId()).isEqualTo(messageId);
|
||||
assertThat(message.getStream()).isEqualTo(key);
|
||||
|
||||
if (!(key instanceof byte[] || value instanceof byte[])) {
|
||||
assertThat(message.getBody()).containsEntry(key, value);
|
||||
}
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void sizeShouldReportStreamSize() {
|
||||
|
||||
K key = keyFactory.instance();
|
||||
V value = valueFactory.instance();
|
||||
|
||||
streamOps.add(key, Collections.singletonMap(key, value));
|
||||
assertThat(streamOps.size(key)).isEqualTo(1);
|
||||
|
||||
streamOps.add(key, Collections.singletonMap(key, value));
|
||||
assertThat(streamOps.size(key)).isEqualTo(2);
|
||||
}
|
||||
}
|
||||
@@ -342,6 +342,11 @@ public class RedisTemplateTests<K, V> {
|
||||
operations.opsForList().rightPop(key1);
|
||||
operations.opsForList().size(key1);
|
||||
operations.exec();
|
||||
|
||||
try {
|
||||
// Await EXEC completion as it's executed on a dedicated connection.
|
||||
Thread.sleep(100);
|
||||
} catch (InterruptedException e) {}
|
||||
operations.opsForValue().set(key1, value1);
|
||||
operations.opsForValue().get(key1);
|
||||
return null;
|
||||
@@ -704,11 +709,7 @@ public class RedisTemplateTests<K, V> {
|
||||
}
|
||||
});
|
||||
|
||||
if (redisTemplate.getConnectionFactory() instanceof JedisConnectionFactory) {
|
||||
assertThat(results, is(empty()));
|
||||
} else {
|
||||
assertNull(results);
|
||||
}
|
||||
assertThat(results, is(empty()));
|
||||
|
||||
assertThat(redisTemplate.opsForValue().get(key1), isEqual(value2));
|
||||
}
|
||||
@@ -777,11 +778,7 @@ public class RedisTemplateTests<K, V> {
|
||||
}
|
||||
});
|
||||
|
||||
if (redisTemplate.getConnectionFactory() instanceof JedisConnectionFactory) {
|
||||
assertThat(results, is(empty()));
|
||||
} else {
|
||||
assertNull(results);
|
||||
}
|
||||
|
||||
assertThat(redisTemplate.opsForValue().get(key1), isEqual(value2));
|
||||
}
|
||||
|
||||
@@ -0,0 +1,97 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.stream;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
|
||||
import java.util.Optional;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link ReadOffsetStrategy}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class ReadOffsetStrategyUnitTests {
|
||||
|
||||
static Optional<Consumer> consumer = Optional.of(Consumer.from("foo", "bar"));
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void nextMessageStandaloneShouldReturnLastSeenMessageId() {
|
||||
|
||||
ReadOffset offset = ReadOffset.from("foo");
|
||||
|
||||
assertThat(ReadOffsetStrategy.NextMessage.getFirst(offset, Optional.empty())).isEqualTo(offset);
|
||||
assertThat(ReadOffsetStrategy.NextMessage.getNext(offset, Optional.empty(), "42")).isEqualTo(ReadOffset.from("42"));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void lastConsumedStandaloneShouldReturnLastSeenMessageId() {
|
||||
|
||||
ReadOffset offset = ReadOffset.lastConsumed();
|
||||
|
||||
assertThat(ReadOffsetStrategy.LastConsumed.getFirst(offset, Optional.empty())).isEqualTo(ReadOffset.latest());
|
||||
assertThat(ReadOffsetStrategy.LastConsumed.getNext(offset, Optional.empty(), "42"))
|
||||
.isEqualTo(ReadOffset.from("42"));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void latestStandaloneShouldReturnLatest() {
|
||||
|
||||
ReadOffset offset = ReadOffset.latest();
|
||||
|
||||
assertThat(ReadOffsetStrategy.Latest.getFirst(offset, Optional.empty())).isEqualTo(ReadOffset.latest());
|
||||
assertThat(ReadOffsetStrategy.Latest.getNext(offset, Optional.empty(), "42")).isEqualTo(ReadOffset.latest());
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void nextMessageConsumerGroupShouldReturnLastSeenMessageId() {
|
||||
|
||||
ReadOffset offset = ReadOffset.from("foo");
|
||||
|
||||
assertThat(ReadOffsetStrategy.NextMessage.getFirst(offset, consumer)).isEqualTo(offset);
|
||||
assertThat(ReadOffsetStrategy.NextMessage.getNext(offset, consumer, "42")).isEqualTo(ReadOffset.from("42"));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void lastConsumedConsumerGroupShouldReturnLastSeenMessageId() {
|
||||
|
||||
ReadOffset offset = ReadOffset.lastConsumed();
|
||||
|
||||
assertThat(ReadOffsetStrategy.LastConsumed.getFirst(offset, consumer)).isEqualTo(ReadOffset.lastConsumed());
|
||||
assertThat(ReadOffsetStrategy.LastConsumed.getNext(offset, consumer, "42")).isEqualTo(ReadOffset.lastConsumed());
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void latestConsumerGroupShouldReturnLatest() {
|
||||
|
||||
ReadOffset offset = ReadOffset.latest();
|
||||
|
||||
assertThat(ReadOffsetStrategy.Latest.getFirst(offset, consumer)).isEqualTo(ReadOffset.latest());
|
||||
assertThat(ReadOffsetStrategy.Latest.getNext(offset, consumer, "42")).isEqualTo(ReadOffset.latest());
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void getStrategyShouldReturnAppropriateStrategy() {
|
||||
|
||||
assertThat(ReadOffsetStrategy.getStrategy(ReadOffset.from("foo"))).isEqualTo(ReadOffsetStrategy.NextMessage);
|
||||
assertThat(ReadOffsetStrategy.getStrategy(ReadOffset.lastConsumed())).isEqualTo(ReadOffsetStrategy.LastConsumed);
|
||||
assertThat(ReadOffsetStrategy.getStrategy(ReadOffset.latest())).isEqualTo(ReadOffsetStrategy.Latest);
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,286 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.stream;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.redis.ConnectionFactoryTracker;
|
||||
import org.springframework.data.redis.RedisVersionUtils;
|
||||
import org.springframework.data.redis.SettingsUtils;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.connection.RedisConnectionFactory;
|
||||
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
|
||||
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest;
|
||||
|
||||
/**
|
||||
* Integration tests for {@link StreamMessageListenerContainer}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
private static final RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(
|
||||
SettingsUtils.getHost(), SettingsUtils.getPort());
|
||||
|
||||
private static RedisConnectionFactory connectionFactory;
|
||||
|
||||
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
StreamMessageListenerContainerOptions<String, String> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().pollTimeout(Duration.ofMillis(100)).build();
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() {
|
||||
|
||||
LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder() //
|
||||
.shutdownTimeout(Duration.ZERO) //
|
||||
.clientResources(LettuceTestClientResources.getSharedClientResources()) //
|
||||
.build();
|
||||
|
||||
LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration,
|
||||
clientConfiguration);
|
||||
lettuceConnectionFactory.afterPropertiesSet();
|
||||
|
||||
ConnectionFactoryTracker.add(lettuceConnectionFactory);
|
||||
|
||||
connectionFactory = lettuceConnectionFactory;
|
||||
|
||||
// TODO: Upgrade to 5.0
|
||||
assumeTrue(RedisVersionUtils.atLeast("4.9", connectionFactory.getConnection()));
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void tearDown() {
|
||||
ConnectionFactoryTracker.cleanUp();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
|
||||
RedisConnection connection = connectionFactory.getConnection();
|
||||
connection.flushDb();
|
||||
connection.close();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessages() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3"));
|
||||
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
|
||||
cancelAwait(subscription);
|
||||
|
||||
assertThat(subscription.isActive()).isFalse();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessagesInConsumerGroup() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
|
||||
String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()), queue::add);
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
|
||||
|
||||
StreamMessage<String, String> message = queue.poll(1, TimeUnit.SECONDS);
|
||||
assertThat(message).isNotNull();
|
||||
assertThat(message.getBody()).containsEntry("key", "value2");
|
||||
|
||||
cancelAwait(subscription);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldUseCustomErrorHandler() throws InterruptedException {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainerOptions<String, String> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().errorHandler(failures::add).pollTimeout(Duration.ofMillis(100)).build();
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(Consumer.from("my-group", "my-consumer"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()), it -> {});
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
Throwable error = failures.poll(1, TimeUnit.SECONDS);
|
||||
assertThat(failures).isEmpty();
|
||||
assertThat(error).isNotNull();
|
||||
|
||||
cancelAwait(subscription);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void errorShouldStopListening() throws InterruptedException {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
StreamReadRequest<String> readRequest = StreamReadRequest
|
||||
.builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())).errorHandler(failures::add)
|
||||
.consumer(Consumer.from("my-group", "my-consumer")).build();
|
||||
|
||||
String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.register(readRequest, it -> {});
|
||||
|
||||
subscription.await(Duration.ofSeconds(1));
|
||||
|
||||
redisTemplate.delete("my-stream");
|
||||
|
||||
subscription.await(Duration.ofSeconds(1));
|
||||
|
||||
assertThat(failures.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(subscription.isActive()).isFalse();
|
||||
|
||||
cancelAwait(subscription);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void customizedCancelPredicateShouldNotStopListening() throws InterruptedException {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
StreamReadRequest<String> readRequest = StreamReadRequest
|
||||
.builder(StreamOffset.create("my-stream", ReadOffset.lastConsumed())) //
|
||||
.errorHandler(failures::add) // //
|
||||
.cancelOnError(t -> false) //
|
||||
.consumer(Consumer.from("my-group", "my-consumer")) //
|
||||
.build();
|
||||
|
||||
String messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.register(readRequest, it -> {});
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
redisTemplate.delete("my-stream");
|
||||
|
||||
assertThat(failures.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(failures.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
assertThat(subscription.isActive()).isTrue();
|
||||
|
||||
cancelAwait(subscription);
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void cancelledStreamShouldNotReceiveMessages() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
cancelAwait(subscription);
|
||||
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value4"));
|
||||
|
||||
assertThat(queue.poll(200, TimeUnit.MILLISECONDS)).isNull();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void containerRestartShouldRestartSubscription() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, String> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<StreamMessage<String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
container.stop();
|
||||
|
||||
while (subscription.isActive()) {
|
||||
Thread.sleep(10);
|
||||
}
|
||||
|
||||
container.start();
|
||||
|
||||
subscription.await(Duration.ofSeconds(2));
|
||||
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull();
|
||||
|
||||
cancelAwait(subscription);
|
||||
}
|
||||
|
||||
private static void cancelAwait(Subscription subscription) throws InterruptedException {
|
||||
|
||||
subscription.cancel();
|
||||
|
||||
while (subscription.isActive()) {
|
||||
Thread.sleep(10);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -0,0 +1,199 @@
|
||||
/*
|
||||
* Copyright 2018 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.redis.stream;
|
||||
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
|
||||
import org.junit.AfterClass;
|
||||
import org.junit.Before;
|
||||
import org.junit.BeforeClass;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.redis.ConnectionFactoryTracker;
|
||||
import org.springframework.data.redis.RedisSystemException;
|
||||
import org.springframework.data.redis.RedisVersionUtils;
|
||||
import org.springframework.data.redis.SettingsUtils;
|
||||
import org.springframework.data.redis.connection.RedisConnection;
|
||||
import org.springframework.data.redis.connection.RedisStandaloneConfiguration;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamMessage;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceClientConfiguration;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceTestClientResources;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptions;
|
||||
|
||||
/**
|
||||
* Integration tests for {@link StreamReceiver}.
|
||||
*
|
||||
* @author Mark Paluch
|
||||
*/
|
||||
public class StreamReceiverIntegrationTests {
|
||||
|
||||
private static final RedisStandaloneConfiguration standaloneConfiguration = new RedisStandaloneConfiguration(
|
||||
SettingsUtils.getHost(), SettingsUtils.getPort());
|
||||
|
||||
private static LettuceConnectionFactory connectionFactory;
|
||||
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() {
|
||||
|
||||
LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder() //
|
||||
.shutdownTimeout(Duration.ZERO) //
|
||||
.clientResources(LettuceTestClientResources.getSharedClientResources()) //
|
||||
.build();
|
||||
|
||||
LettuceConnectionFactory lettuceConnectionFactory = new LettuceConnectionFactory(standaloneConfiguration,
|
||||
clientConfiguration);
|
||||
lettuceConnectionFactory.afterPropertiesSet();
|
||||
|
||||
ConnectionFactoryTracker.add(lettuceConnectionFactory);
|
||||
|
||||
connectionFactory = lettuceConnectionFactory;
|
||||
|
||||
// TODO: Upgrade to 5.0
|
||||
assumeTrue(RedisVersionUtils.atLeast("4.9", connectionFactory.getConnection()));
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
public static void tearDown() {
|
||||
ConnectionFactoryTracker.cleanUp();
|
||||
}
|
||||
|
||||
@Before
|
||||
public void before() {
|
||||
|
||||
RedisConnection connection = connectionFactory.getConnection();
|
||||
connection.flushDb();
|
||||
connection.close();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessages() {
|
||||
|
||||
StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory);
|
||||
|
||||
Flux<StreamMessage<String, String>> messages = receiver
|
||||
.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")));
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.then(() -> redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value")))
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
assertThat(it.getBody()).containsEntry("key", "value");
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void latestModeLosesMessages() {
|
||||
|
||||
// XADD/XREAD highly timing-dependent as this tests require a poll subscription to receive messages using $ offset.
|
||||
|
||||
StreamReceiverOptions<String, String> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofSeconds(4))
|
||||
.build();
|
||||
StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory, options);
|
||||
|
||||
Flux<StreamMessage<String, String>> messages = receiver
|
||||
.receive(StreamOffset.create("my-stream", ReadOffset.latest()));
|
||||
|
||||
messages.as(publisher -> StepVerifier.create(publisher, 0)) //
|
||||
.thenRequest(1) //
|
||||
.then(() -> {
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
} catch (InterruptedException e) {}
|
||||
}) //
|
||||
.expectNextCount(1) //
|
||||
.then(() -> {
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
|
||||
}) //
|
||||
.thenRequest(1) //
|
||||
.then(() -> {
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3"));
|
||||
} catch (InterruptedException e) {}
|
||||
}).consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
assertThat(it.getBody()).containsEntry("key", "value3");
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveAsConsumerGroupMessages() {
|
||||
|
||||
StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory);
|
||||
|
||||
Flux<StreamMessage<String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
|
||||
|
||||
// required to initialize stream
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group");
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key2", "value2"));
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
assertThat(it.getBody()).containsEntry("key", "value");
|
||||
}).consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
assertThat(it.getBody()).containsEntry("key2", "value2");
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldStopReceivingOnError() {
|
||||
|
||||
StreamReceiverOptions<String, String> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofMillis(100))
|
||||
.build();
|
||||
|
||||
StreamReceiver<String, String> receiver = StreamReceiver.create(connectionFactory, options);
|
||||
|
||||
Flux<StreamMessage<String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
|
||||
|
||||
// required to initialize stream
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from("0-0"), "my-group");
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.then(() -> redisTemplate.delete("my-stream")) //
|
||||
.expectError(RedisSystemException.class) //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user