DATAREDIS-864 - Polishing.
Exract nested Stream types into connection.stream package. Extract common code from Template-API based toObjectRecord handling into StreamObjectMapper. StreamReceiver and StreamMessageListenerContainer now can emit MapRecord or ObjectRecord for easier consumption of stream records. Original Pull Request: #356
This commit is contained in:
Mark Paluch
committed by
Christoph Strobl
Christoph Strobl
parent
13648d7c85
commit
eafcb331d8
@@ -64,11 +64,11 @@ 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.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.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;
|
||||
|
||||
@@ -26,7 +26,8 @@ import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.geo.Distance;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
|
||||
/**
|
||||
* Unit test of {@link DefaultStringRedisConnection} that executes commands in a pipeline
|
||||
|
||||
@@ -27,7 +27,8 @@ import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.geo.Distance;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
|
||||
/**
|
||||
* @author Jennifer Hickey
|
||||
|
||||
@@ -45,11 +45,11 @@ import org.springframework.data.redis.connection.RedisGeoCommands.GeoLocation;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands.GeoRadiusCommandArgs;
|
||||
import org.springframework.data.redis.connection.RedisListCommands.Position;
|
||||
import org.springframework.data.redis.connection.RedisServerCommands.ShutdownOption;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.RedisStringCommands.BitOperation;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Aggregate;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
@@ -57,6 +57,7 @@ import org.springframework.data.redis.connection.RedisZSetCommands.Tuple;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Weights;
|
||||
import org.springframework.data.redis.connection.StringRedisConnection.StringTuple;
|
||||
import org.springframework.data.redis.connection.convert.Converters;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
|
||||
/**
|
||||
|
||||
@@ -27,7 +27,8 @@ import org.junit.Ignore;
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.geo.Distance;
|
||||
import org.springframework.data.redis.connection.RedisGeoCommands.DistanceUnit;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
|
||||
/**
|
||||
* @author Jennifer Hickey
|
||||
|
||||
@@ -21,11 +21,12 @@ import java.util.Collections;
|
||||
import java.util.Map;
|
||||
|
||||
import org.junit.Test;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ByteRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Record;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.ByteRecord;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.stream.Record;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
import org.springframework.data.redis.hash.HashMapper;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
|
||||
|
||||
@@ -1,405 +0,0 @@
|
||||
package org.springframework.data.redis.connection.lettuce;
|
||||
|
||||
import io.lettuce.core.Limit;
|
||||
import io.lettuce.core.RedisClient;
|
||||
import io.lettuce.core.RedisURI;
|
||||
import io.lettuce.core.StreamMessage;
|
||||
import io.lettuce.core.api.StatefulRedisConnection;
|
||||
import lombok.Data;
|
||||
|
||||
import java.nio.charset.StandardCharsets;
|
||||
import java.util.Collections;
|
||||
import java.util.List;
|
||||
import java.util.Map;
|
||||
import java.util.stream.Collectors;
|
||||
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
import org.springframework.core.convert.support.DefaultConversionService;
|
||||
import org.springframework.data.domain.Range;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Record;
|
||||
import org.springframework.data.redis.connection.StreamRecords;
|
||||
import org.springframework.data.redis.core.convert.RedisCustomConversions;
|
||||
import org.springframework.data.redis.hash.HashMapper;
|
||||
import org.springframework.data.redis.hash.Jackson2HashMapper;
|
||||
import org.springframework.data.redis.hash.ObjectHashMapper;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.util.ClassUtils;
|
||||
|
||||
/**
|
||||
* @author Christoph Strobl
|
||||
* @since 2018/10
|
||||
*/
|
||||
public class ApiSpike {
|
||||
|
||||
RedisClient client;
|
||||
StatefulRedisConnection<byte[], byte[]> connection;
|
||||
|
||||
LettuceConnection lc;
|
||||
|
||||
@Before
|
||||
public void setUp() {
|
||||
|
||||
client = RedisClient.create(RedisURI.create("localhost", 6379));
|
||||
lc = new LettuceConnection(1, client);
|
||||
}
|
||||
|
||||
@After
|
||||
public void tearDown() {
|
||||
|
||||
lc.flushAll();
|
||||
lc.close();
|
||||
client.shutdown();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void all() {
|
||||
|
||||
plainStuff();
|
||||
System.out.println("-----------");
|
||||
simpleObject();
|
||||
System.out.println("-----------");
|
||||
writeWithHashReadWithMap();
|
||||
System.out.println("-----------");
|
||||
}
|
||||
|
||||
@Test
|
||||
public void plainStuff() {
|
||||
|
||||
RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
|
||||
|
||||
StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
|
||||
RedisSerializer.string(), RedisSerializer.java(), null);
|
||||
|
||||
RecordId id = ops.xAdd("foo", Record.of(Collections.singletonMap("field", "value")));
|
||||
List<MapRecord<String, String, Object>> range = ops.xRange("foo", Range.unbounded());
|
||||
|
||||
range.forEach(it -> System.out.println(it.getId() + ": " + it.getValue()));
|
||||
|
||||
List<ObjectRecord<String, String>> stringRange = ops.xRange("foo", Range.unbounded(), String.class);
|
||||
stringRange.forEach(System.out::println);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void simpleObject() {
|
||||
|
||||
SimpleObject o = new SimpleObject();
|
||||
o.field1 = "value-1";
|
||||
o.field2 = 10L;
|
||||
|
||||
RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
|
||||
|
||||
StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
|
||||
RedisSerializer.string(), RedisSerializer.java(), new Jackson2HashMapper(false));
|
||||
|
||||
RecordId id = ops.xAdd("key", o);
|
||||
List<ObjectRecord<String, SimpleObject>> list = ops.xRange("key", Range.unbounded(), SimpleObject.class);
|
||||
|
||||
list.forEach(System.out::println);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void writeWithHashReadWithMap() {
|
||||
|
||||
SimpleObject o = new SimpleObject();
|
||||
o.field1 = "value-1";
|
||||
o.field2 = 10L;
|
||||
|
||||
RedisStreamCommandsImpl imp = new RedisStreamCommandsImpl(lc);
|
||||
|
||||
StreamOperationsImpl<String, String, Object> ops = new StreamOperationsImpl<>(imp, RedisSerializer.string(),
|
||||
RedisSerializer.string(), RedisSerializer.java(), null);
|
||||
|
||||
RecordId id = ops.xAdd("key", o);
|
||||
|
||||
List<MapRecord<String, String, Object>> list = ops.xRange("key", Range.unbounded());
|
||||
|
||||
list.forEach(System.out::println);
|
||||
}
|
||||
|
||||
@Data
|
||||
static class SimpleObject {
|
||||
|
||||
String field1;
|
||||
Long field2;
|
||||
}
|
||||
|
||||
@Test
|
||||
public void x2() {
|
||||
|
||||
RedisCustomConversions rcc = new RedisCustomConversions();
|
||||
DefaultConversionService conversionService = new DefaultConversionService();
|
||||
rcc.registerConvertersIn(conversionService);
|
||||
|
||||
}
|
||||
|
||||
interface RedisStreamCommands {
|
||||
|
||||
RecordId xAdd(byte[] key, MapRecord<byte[], byte[], byte[]> entry);
|
||||
|
||||
List<MapRecord<byte[], byte[], byte[]>> xRange(byte[] key, Range<String> range);
|
||||
|
||||
}
|
||||
|
||||
static class RedisStreamCommandsImpl implements RedisStreamCommands {
|
||||
|
||||
private LettuceConnection connection;
|
||||
|
||||
public RedisStreamCommandsImpl(LettuceConnection connection) {
|
||||
this.connection = connection;
|
||||
}
|
||||
|
||||
@Override
|
||||
public RecordId xAdd(byte[] key, MapRecord<byte[],byte[], byte[]> entry) {
|
||||
return RecordId
|
||||
.of(connection.getConnection().xadd(key, entry.getValue()));
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<MapRecord<byte[], byte[], byte[]>> xRange(byte[] key, Range<String> range) {
|
||||
|
||||
List<StreamMessage<byte[], byte[]>> raw = connection.getConnection().xrange(key,
|
||||
io.lettuce.core.Range.unbounded(), Limit.unlimited());
|
||||
return raw.stream()
|
||||
.map(it -> StreamRecords.rawBytes(it.getBody())
|
||||
.withId(RecordId.of(it.getId())))
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
}
|
||||
|
||||
interface StreamOperations<K, HK, HV> {
|
||||
|
||||
default RecordId xAdd(K key, Object value) {
|
||||
return xAdd(key, Record.of(value));
|
||||
}
|
||||
|
||||
default RecordId xAdd(K key, Record<K,?> value) {
|
||||
return xAdd(key, objectToEntry(value));
|
||||
}
|
||||
|
||||
RecordId xAdd(K key, MapRecord<K, ? extends HK, ? extends HV> entry);
|
||||
|
||||
List<MapRecord<K, HK, HV>> xRange(K key, Range<String> range);
|
||||
|
||||
default <V> List<ObjectRecord<K, V>> xRange(K key, Range<String> range, Class<V> targetType) {
|
||||
return xRange(key, range).stream().map(it -> entryToObject(it, targetType)).collect(Collectors.toList());
|
||||
}
|
||||
|
||||
default <V> MapRecord<K, HK, HV> objectToEntry(V value) {
|
||||
|
||||
if (value instanceof ObjectRecord) {
|
||||
|
||||
ObjectRecord entry = ((ObjectRecord) value);
|
||||
return Record.of(((HashMapper) getHashMapper(entry.getValue().getClass())).toHash(entry.getValue()))
|
||||
.withId(entry.getId());
|
||||
}
|
||||
|
||||
return Record.of(((HashMapper) getHashMapper(value.getClass())).toHash(value));
|
||||
}
|
||||
|
||||
default <V> ObjectRecord<K, V> entryToObject(MapRecord<K, HK, HV> entry, Class<V> targetType) {
|
||||
return entry.toObjectRecord(getHashMapper(targetType));
|
||||
}
|
||||
|
||||
<V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType);
|
||||
}
|
||||
|
||||
/*
|
||||
* Conversion Rules
|
||||
*
|
||||
* 1) Simple types
|
||||
* serialize: default value serializer (key known byte array of class)
|
||||
* deserialized: default value serializer
|
||||
* 2) Complex types
|
||||
* serialize: HashMapper: check if all entries are binary - then pass on to serializer
|
||||
* deserialize: deserialize then pass to hashMapper
|
||||
*
|
||||
*
|
||||
*
|
||||
*
|
||||
*/
|
||||
static class StreamOperationsImpl<K, HK, HV> implements StreamOperations<K, HK, HV> {
|
||||
|
||||
private RedisSerializer<K> keySerializer;
|
||||
private RedisSerializer hashKeySerializer;
|
||||
private RedisSerializer hashValueSerializer;
|
||||
private RedisStreamCommands commands;
|
||||
private final RedisCustomConversions rcc = new RedisCustomConversions();
|
||||
private DefaultConversionService conversionService;
|
||||
|
||||
private HashMapper<?, HK, HV> mapper;
|
||||
|
||||
public StreamOperationsImpl(RedisStreamCommands commands, RedisSerializer<K> keySerializer,
|
||||
RedisSerializer<HK> hashKeySerializer, RedisSerializer<HV> hashValueSerializer, HashMapper<?, HK, HV> mapper) {
|
||||
|
||||
this.commands = commands;
|
||||
this.keySerializer = keySerializer;
|
||||
this.hashKeySerializer = hashKeySerializer;
|
||||
this.hashValueSerializer = hashValueSerializer;
|
||||
|
||||
this.conversionService = new DefaultConversionService();
|
||||
this.mapper = mapper != null ? mapper : (HashMapper<?, HK, HV>) new ObjectHashMapper();
|
||||
rcc.registerConvertersIn(conversionService);
|
||||
}
|
||||
|
||||
@Override
|
||||
public RecordId xAdd(K key, MapRecord<K, ? extends HK, ? extends HV> entry) {
|
||||
return commands.xAdd(serializeKeyIfRequired(key), entry.mapEntries(this::mapToBinary).withStreamKey(serializeKeyIfRequired(key)));
|
||||
}
|
||||
|
||||
@Override
|
||||
public List<MapRecord<K, HK, HV>> xRange(K key, Range<String> range) {
|
||||
|
||||
return commands.xRange(serializeKeyIfRequired(key), range).stream().map(it ->it.mapEntries(this::mapToObject).withStreamKey(deserializeKey(it.getStream(), null)))
|
||||
.collect(Collectors.toList());
|
||||
}
|
||||
|
||||
@Override
|
||||
public <V> HashMapper<V, HK, HV> getHashMapper(Class<V> targetType) {
|
||||
|
||||
if (rcc.isSimpleType(targetType)) {
|
||||
|
||||
return new HashMapper<V, HK, HV>() {
|
||||
|
||||
@Override
|
||||
public Map<HK, HV> toHash(V object) {
|
||||
return (Map<HK, HV>) Collections.singletonMap("payload".getBytes(StandardCharsets.UTF_8),
|
||||
serializeHashValueIfRequires((HV) object));
|
||||
}
|
||||
|
||||
@Override
|
||||
public V fromHash(Map<HK, HV> hash) {
|
||||
Object value = hash.values().iterator().next();
|
||||
if (ClassUtils.isAssignableValue(targetType, value)) {
|
||||
return (V) value;
|
||||
}
|
||||
return (V) deserializeHashValue((byte[]) value, (Class<HV>) targetType);
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
if (mapper instanceof ObjectHashMapper) {
|
||||
|
||||
return new HashMapper<V, HK, HV>() {
|
||||
|
||||
@Override
|
||||
public Map<HK, HV> toHash(V object) {
|
||||
return (Map<HK, HV>) ((ObjectHashMapper) mapper).toObjectHash(object);
|
||||
}
|
||||
|
||||
@Override
|
||||
public V fromHash(Map<HK, HV> hash) {
|
||||
|
||||
Map<byte[], byte[]> map = hash.entrySet().stream()
|
||||
.collect(Collectors.toMap(e -> conversionService.convert((Object) e.getKey(), byte[].class),
|
||||
e -> conversionService.convert((Object) e.getValue(), byte[].class)));
|
||||
|
||||
return (V) mapper.fromHash((Map<HK, HV>) map);
|
||||
}
|
||||
};
|
||||
|
||||
}
|
||||
|
||||
return (HashMapper<V, HK, HV>) mapper;
|
||||
}
|
||||
|
||||
protected byte[] serializeHashKeyIfRequired(HK key) {
|
||||
|
||||
return hashKeySerializerPresent() ? serialize(key, hashKeySerializer)
|
||||
: conversionService.convert(key, byte[].class);
|
||||
}
|
||||
|
||||
protected boolean hashKeySerializerPresent() {
|
||||
return hashValueSerializer != null;
|
||||
}
|
||||
|
||||
protected byte[] serializeHashValueIfRequires(HV value) {
|
||||
return hashValueSerializerPresent() ? serialize(value, hashValueSerializer)
|
||||
: conversionService.convert(value, byte[].class);
|
||||
}
|
||||
|
||||
protected boolean hashValueSerializerPresent() {
|
||||
return hashValueSerializer != null;
|
||||
}
|
||||
|
||||
protected byte[] serializeKeyIfRequired(K key) {
|
||||
return keySerializerPresent() ? serialize(key, keySerializer) : conversionService.convert(key, byte[].class);
|
||||
}
|
||||
|
||||
protected boolean keySerializerPresent() {
|
||||
return keySerializer != null;
|
||||
}
|
||||
|
||||
protected K deserializeKey(byte[] bytes, Class<K> targetType) {
|
||||
return keySerializerPresent() ? keySerializer.deserialize(bytes) : conversionService.convert(bytes, targetType);
|
||||
}
|
||||
|
||||
protected HK deserializeHashKey(byte[] bytes, Class<HK> targetType) {
|
||||
|
||||
return hashKeySerializerPresent() ? (HK) hashKeySerializer.deserialize(bytes)
|
||||
: conversionService.convert(bytes, targetType);
|
||||
}
|
||||
|
||||
protected HV deserializeHashValue(byte[] bytes, Class<HV> targetType) {
|
||||
return hashValueSerializerPresent() ? (HV) hashValueSerializer.deserialize(bytes)
|
||||
: conversionService.convert(bytes, targetType);
|
||||
}
|
||||
|
||||
byte[] serialize(Object value, RedisSerializer serializer) {
|
||||
|
||||
Object _value = value;
|
||||
if (!serializer.canSerialize(value.getClass())) {
|
||||
_value = conversionService.convert(value, serializer.getTargetType());
|
||||
}
|
||||
return serializer.serialize(_value);
|
||||
}
|
||||
|
||||
private Map.Entry<byte[], byte[]> mapToBinary(Map.Entry<? extends HK, ? extends HV> it) {
|
||||
|
||||
return new Map.Entry<byte[], byte[]>() {
|
||||
|
||||
@Override
|
||||
public byte[] getKey() {
|
||||
return serializeHashKeyIfRequired(it.getKey());
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] getValue() {
|
||||
return serializeHashValueIfRequires(it.getValue());
|
||||
}
|
||||
|
||||
@Override
|
||||
public byte[] setValue(byte[] value) {
|
||||
return new byte[0];
|
||||
}
|
||||
};
|
||||
}
|
||||
|
||||
private Map.Entry<HK, HV> mapToObject(Map.Entry<byte[], byte[]> pair) {
|
||||
|
||||
return new Map.Entry<HK, HV>() {
|
||||
|
||||
@Override
|
||||
public HK getKey() {
|
||||
return deserializeHashKey(pair.getKey(), (Class<HK>) Object.class);
|
||||
}
|
||||
|
||||
@Override
|
||||
public HV getValue() {
|
||||
return deserializeHashValue(pair.getValue(), (Class<HV>) Object.class);
|
||||
}
|
||||
|
||||
@Override
|
||||
public HV setValue(HV value) {
|
||||
return value;
|
||||
}
|
||||
|
||||
};
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
@@ -20,7 +20,7 @@ import static org.junit.Assume.*;
|
||||
|
||||
import io.lettuce.core.XReadArgs;
|
||||
import org.junit.Ignore;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
import java.util.Collections;
|
||||
@@ -29,9 +29,9 @@ 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.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
|
||||
/**
|
||||
|
||||
@@ -33,23 +33,27 @@ 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.Person;
|
||||
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.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
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.StreamRecords;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
import org.springframework.data.redis.serializer.GenericJackson2JsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.Jackson2JsonRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.JdkSerializationRedisSerializer;
|
||||
import org.springframework.data.redis.serializer.OxmSerializer;
|
||||
import org.springframework.data.redis.serializer.RedisSerializationContext;
|
||||
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
|
||||
import org.springframework.data.redis.serializer.RedisSerializer;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
|
||||
/**
|
||||
* Integration tests for {@link DefaultReactiveStreamOperations}.
|
||||
@@ -93,8 +97,7 @@ public class DefaultReactiveStreamOperationsTests<K, HK, HV> {
|
||||
// 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"));
|
||||
assumeTrue(RedisTestProfileValueSource.matches("redisVersion", "5.0"));
|
||||
|
||||
RedisSerializationContext<K, ?> context = null;
|
||||
if (serializer != null) {
|
||||
@@ -150,14 +153,15 @@ public class DefaultReactiveStreamOperationsTests<K, HK, HV> {
|
||||
public void addShouldAddReadSimpleMessage() {
|
||||
|
||||
assumeTrue(!(serializer instanceof Jackson2JsonRedisSerializer)
|
||||
&& !(serializer instanceof GenericJackson2JsonRedisSerializer));
|
||||
&& !(serializer instanceof GenericJackson2JsonRedisSerializer)
|
||||
&& !(serializer instanceof JdkSerializationRedisSerializer) && !(serializer instanceof OxmSerializer));
|
||||
|
||||
K key = keyFactory.instance();
|
||||
HV value = valueFactory.instance();
|
||||
|
||||
RecordId messageId = streamOperations.add(StreamRecords.objectBacked(value).withStreamKey(key)).block();
|
||||
|
||||
streamOperations.range(key, Range.unbounded(), (Class<HV>) value.getClass()).as(StepVerifier::create) //
|
||||
streamOperations.range((Class<HV>) value.getClass(), key, Range.unbounded()).as(StepVerifier::create) //
|
||||
.consumeNextWith(it -> {
|
||||
assertThat(it.getId()).isEqualTo(messageId);
|
||||
assertThat(it.getStream()).isEqualTo(key);
|
||||
@@ -168,6 +172,36 @@ public class DefaultReactiveStreamOperationsTests<K, HK, HV> {
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void addShouldAddReadSimpleMessageWithRawSerializer() {
|
||||
|
||||
assumeTrue(!(serializer instanceof Jackson2JsonRedisSerializer)
|
||||
&& !(serializer instanceof GenericJackson2JsonRedisSerializer));
|
||||
|
||||
SerializationPair<K> keySerializer = redisTemplate.getSerializationContext().getKeySerializationPair();
|
||||
|
||||
RedisSerializationContext<K, String> serializationContext = RedisSerializationContext
|
||||
.<K, String> newSerializationContext(StringRedisSerializer.UTF_8).key(keySerializer)
|
||||
.hashValue(SerializationPair.raw()).hashKey(SerializationPair.raw()).build();
|
||||
|
||||
ReactiveRedisTemplate<K, String> raw = new ReactiveRedisTemplate<>(redisTemplate.getConnectionFactory(),
|
||||
serializationContext);
|
||||
|
||||
K key = keyFactory.instance();
|
||||
Person value = new PersonObjectFactory().instance();
|
||||
|
||||
RecordId messageId = raw.opsForStream().add(StreamRecords.objectBacked(value).withStreamKey(key)).block();
|
||||
|
||||
raw.opsForStream().range((Class<HV>) value.getClass(), key, Range.unbounded()).as(StepVerifier::create) //
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getId()).isEqualTo(messageId);
|
||||
assertThat(it.getStream()).isEqualTo(key);
|
||||
assertThat(it.getValue()).isEqualTo(value);
|
||||
}) //
|
||||
.verifyComplete();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void rangeShouldReportMessages() {
|
||||
|
||||
|
||||
@@ -35,21 +35,22 @@ import org.springframework.data.redis.ConnectionFactoryTracker;
|
||||
import org.springframework.data.redis.ObjectFactory;
|
||||
import org.springframework.data.redis.Person;
|
||||
import org.springframework.data.redis.RedisTestProfileValueSource;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Consumer;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.RecordId;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamReadOptions;
|
||||
import org.springframework.data.redis.connection.RedisZSetCommands.Limit;
|
||||
import org.springframework.data.redis.connection.StreamRecords;
|
||||
import org.springframework.data.redis.connection.stream.StreamRecords;
|
||||
import org.springframework.data.redis.connection.lettuce.LettuceConnectionFactory;
|
||||
|
||||
/**
|
||||
* Integration test of {@link DefaultStreamOperations}
|
||||
*
|
||||
* @author Mark Paluch
|
||||
* @author Christoph Strobl
|
||||
*/
|
||||
@RunWith(Parameterized.class)
|
||||
public class DefaultStreamOperationsTests<K, HK, HV> {
|
||||
@@ -71,7 +72,6 @@ public class DefaultStreamOperationsTests<K, HK, HV> {
|
||||
// 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", "5.0"));
|
||||
|
||||
this.redisTemplate = redisTemplate;
|
||||
@@ -133,7 +133,7 @@ public class DefaultStreamOperationsTests<K, HK, HV> {
|
||||
|
||||
RecordId messageId = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
|
||||
List<ObjectRecord<K, HV>> messages = streamOps.range(key, Range.unbounded(), (Class<HV>) value.getClass());
|
||||
List<ObjectRecord<K, HV>> messages = streamOps.range((Class<HV>) value.getClass(), key, Range.unbounded());
|
||||
|
||||
assertThat(messages).hasSize(1);
|
||||
|
||||
@@ -212,7 +212,7 @@ public class DefaultStreamOperationsTests<K, HK, HV> {
|
||||
RecordId messageId1 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
RecordId messageId2 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
|
||||
List<ObjectRecord<K, HV>> messages = streamOps.reverseRange(key, Range.unbounded(), (Class<HV>) value.getClass());
|
||||
List<ObjectRecord<K, HV>> messages = streamOps.reverseRange((Class<HV>) value.getClass(), key, Range.unbounded());
|
||||
|
||||
assertThat(messages).hasSize(2).extracting("id").containsSequence(messageId2, messageId1);
|
||||
|
||||
@@ -254,7 +254,7 @@ public class DefaultStreamOperationsTests<K, HK, HV> {
|
||||
HV value = hashValueFactory.instance();
|
||||
|
||||
RecordId messageId1 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
RecordId messageId2 = streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
streamOps.add(StreamRecords.objectBacked(value).withStreamKey(key));
|
||||
|
||||
List<ObjectRecord<K, HV>> messages = streamOps.read((Class<HV>) value.getClass(), StreamOffset.create(key, ReadOffset.from("0-0")));
|
||||
|
||||
|
||||
@@ -20,8 +20,8 @@ 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;
|
||||
import org.springframework.data.redis.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
|
||||
/**
|
||||
* Unit tests for {@link ReadOffsetStrategy}.
|
||||
|
||||
@@ -18,9 +18,11 @@ package org.springframework.data.redis.stream;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
|
||||
import java.time.Duration;
|
||||
import java.util.Collections;
|
||||
import java.util.Map;
|
||||
import java.util.concurrent.BlockingQueue;
|
||||
import java.util.concurrent.LinkedBlockingQueue;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
@@ -35,14 +37,16 @@ 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.RecordId;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.Record;
|
||||
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.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.Record;
|
||||
import org.springframework.data.redis.connection.stream.RecordId;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamMessageListenerContainerOptions;
|
||||
import org.springframework.data.redis.stream.StreamMessageListenerContainer.StreamReadRequest;
|
||||
@@ -59,8 +63,8 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
private static RedisConnectionFactory connectionFactory;
|
||||
|
||||
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
StreamMessageListenerContainerOptions<String, Map<String,String>> containerOptions = StreamMessageListenerContainerOptions
|
||||
private StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
private StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().pollTimeout(Duration.ofMillis(100)).build();
|
||||
|
||||
@BeforeClass
|
||||
@@ -79,8 +83,7 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
connectionFactory = lettuceConnectionFactory;
|
||||
|
||||
// TODO: Upgrade to 5.0
|
||||
assumeTrue(RedisVersionUtils.atLeast("4.9", connectionFactory.getConnection()));
|
||||
assumeTrue(RedisVersionUtils.atLeast("5.0", connectionFactory.getConnection()));
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
@@ -97,11 +100,12 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessages() throws InterruptedException {
|
||||
public void shouldReceiveMapMessages() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String,String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<Record<String, Map<String, String>>> queue = new LinkedBlockingQueue<>();
|
||||
BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
@@ -121,12 +125,62 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
assertThat(subscription.isActive()).isFalse();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveSimpleObjectHashRecords() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainerOptions<String, ObjectRecord<String, String>> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().pollTimeout(Duration.ofMillis(100)).targetType(String.class).build();
|
||||
|
||||
StreamMessageListenerContainer<String, ObjectRecord<String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory, containerOptions);
|
||||
BlockingQueue<ObjectRecord<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(ObjectRecord.create("my-stream", "value1"));
|
||||
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull().extracting(Record::getValue).isEqualTo("value1");
|
||||
|
||||
cancelAwait(subscription);
|
||||
|
||||
assertThat(subscription.isActive()).isFalse();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveObjectHashRecords() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainerOptions<String, ObjectRecord<String, LoginEvent>> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().pollTimeout(Duration.ofMillis(100)).targetType(LoginEvent.class).build();
|
||||
|
||||
StreamMessageListenerContainer<String, ObjectRecord<String, LoginEvent>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory, containerOptions);
|
||||
BlockingQueue<ObjectRecord<String, LoginEvent>> 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(ObjectRecord.create("my-stream", new LoginEvent("Walter", "White")));
|
||||
|
||||
assertThat(queue.poll(1, TimeUnit.SECONDS)).isNotNull().extracting(Record::getValue)
|
||||
.isEqualTo(new LoginEvent("Walter", "White"));
|
||||
|
||||
cancelAwait(subscription);
|
||||
|
||||
assertThat(subscription.isActive()).isFalse();
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessagesInConsumerGroup() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<Record<String, Map<String, String>>> queue = new LinkedBlockingQueue<>();
|
||||
BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
|
||||
RecordId messageId = redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
redisTemplate.opsForStream().createGroup("my-stream", ReadOffset.from(messageId), "my-group");
|
||||
|
||||
@@ -138,7 +192,7 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
|
||||
|
||||
Record<String, Map<String, String>> message = queue.poll(1, TimeUnit.SECONDS);
|
||||
MapRecord<String, String, String> message = queue.poll(1, TimeUnit.SECONDS);
|
||||
assertThat(message).isNotNull();
|
||||
assertThat(message.getValue()).containsEntry("key", "value2");
|
||||
|
||||
@@ -150,9 +204,10 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainerOptions<String, Map<String, String>> containerOptions = StreamMessageListenerContainerOptions
|
||||
StreamMessageListenerContainerOptions<String, MapRecord<String, String, String>> containerOptions = StreamMessageListenerContainerOptions
|
||||
.builder().errorHandler(failures::add).pollTimeout(Duration.ofMillis(100)).build();
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
container.start();
|
||||
@@ -173,7 +228,8 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
StreamReadRequest<String> readRequest = StreamReadRequest
|
||||
@@ -203,7 +259,8 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
|
||||
BlockingQueue<Throwable> failures = new LinkedBlockingQueue<>();
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
|
||||
StreamReadRequest<String> readRequest = StreamReadRequest
|
||||
@@ -233,9 +290,10 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
@Test // DATAREDIS-864
|
||||
public void cancelledStreamShouldNotReceiveMessages() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<Record<String,Map<String, String>>> queue = new LinkedBlockingQueue<>();
|
||||
BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
@@ -251,9 +309,10 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
@Test // DATAREDIS-864
|
||||
public void containerRestartShouldRestartSubscription() throws InterruptedException {
|
||||
|
||||
StreamMessageListenerContainer<String, Map<String, String>> container = StreamMessageListenerContainer.create(connectionFactory,
|
||||
StreamMessageListenerContainer<String, MapRecord<String, String, String>> container = StreamMessageListenerContainer
|
||||
.create(connectionFactory,
|
||||
containerOptions);
|
||||
BlockingQueue<Record<String,Map<String, String>>> queue = new LinkedBlockingQueue<>();
|
||||
BlockingQueue<MapRecord<String, String, String>> queue = new LinkedBlockingQueue<>();
|
||||
|
||||
container.start();
|
||||
Subscription subscription = container.receive(StreamOffset.create("my-stream", ReadOffset.from("0-0")), queue::add);
|
||||
@@ -285,4 +344,10 @@ public class StreamMessageListenerContainerIntegrationTests {
|
||||
Thread.sleep(10);
|
||||
}
|
||||
}
|
||||
|
||||
@Data
|
||||
@AllArgsConstructor
|
||||
static class LoginEvent {
|
||||
String firstname, lastname;
|
||||
}
|
||||
}
|
||||
|
||||
@@ -18,6 +18,8 @@ package org.springframework.data.redis.stream;
|
||||
import static org.assertj.core.api.Assertions.*;
|
||||
import static org.junit.Assume.*;
|
||||
|
||||
import lombok.AllArgsConstructor;
|
||||
import lombok.Data;
|
||||
import reactor.core.publisher.Flux;
|
||||
import reactor.test.StepVerifier;
|
||||
|
||||
@@ -34,14 +36,19 @@ 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.MapRecord;
|
||||
import org.springframework.data.redis.connection.RedisStreamCommands.ReadOffset;
|
||||
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.connection.stream.Consumer;
|
||||
import org.springframework.data.redis.connection.stream.MapRecord;
|
||||
import org.springframework.data.redis.connection.stream.ObjectRecord;
|
||||
import org.springframework.data.redis.connection.stream.ReadOffset;
|
||||
import org.springframework.data.redis.connection.stream.StreamOffset;
|
||||
import org.springframework.data.redis.core.ReactiveRedisTemplate;
|
||||
import org.springframework.data.redis.core.StringRedisTemplate;
|
||||
import org.springframework.data.redis.serializer.RedisSerializationContext;
|
||||
import org.springframework.data.redis.serializer.RedisSerializationContext.SerializationPair;
|
||||
import org.springframework.data.redis.serializer.StringRedisSerializer;
|
||||
import org.springframework.data.redis.stream.StreamReceiver.StreamReceiverOptions;
|
||||
|
||||
/**
|
||||
@@ -55,7 +62,8 @@ public class StreamReceiverIntegrationTests {
|
||||
SettingsUtils.getHost(), SettingsUtils.getPort());
|
||||
|
||||
private static LettuceConnectionFactory connectionFactory;
|
||||
StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
private StringRedisTemplate redisTemplate = new StringRedisTemplate(connectionFactory);
|
||||
private static ReactiveRedisTemplate<String, String> reactiveRedisTemplate;
|
||||
|
||||
@BeforeClass
|
||||
public static void beforeClass() {
|
||||
@@ -73,8 +81,13 @@ public class StreamReceiverIntegrationTests {
|
||||
|
||||
connectionFactory = lettuceConnectionFactory;
|
||||
|
||||
// TODO: Upgrade to 5.0
|
||||
assumeTrue(RedisVersionUtils.atLeast("5.0", connectionFactory.getConnection()));
|
||||
|
||||
RedisSerializationContext<String, String> serializationContext = RedisSerializationContext
|
||||
.<String, String> newSerializationContext(StringRedisSerializer.UTF_8).hashKey(SerializationPair.raw())
|
||||
.hashValue(SerializationPair.raw()).build();
|
||||
|
||||
reactiveRedisTemplate = new ReactiveRedisTemplate<>(connectionFactory, serializationContext);
|
||||
}
|
||||
|
||||
@AfterClass
|
||||
@@ -91,19 +104,65 @@ public class StreamReceiverIntegrationTests {
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveMessages() {
|
||||
public void shouldReceiveMapRecords() {
|
||||
|
||||
StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory);
|
||||
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory);
|
||||
|
||||
Flux<MapRecord<String, 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")))
|
||||
.then(() -> reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value"))
|
||||
.subscribe())
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
// assertThat(it.getValue()).containsEntry("key", "value");
|
||||
assertThat(it.getValue()).containsEntry("key", "value");
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveSimpleObjectHashRecords() {
|
||||
|
||||
StreamReceiverOptions<String, ObjectRecord<String, String>> receiverOptions = StreamReceiverOptions.builder()
|
||||
.targetType(String.class).build();
|
||||
|
||||
StreamReceiver<String, ObjectRecord<String, String>> receiver = StreamReceiver.create(connectionFactory,
|
||||
receiverOptions);
|
||||
|
||||
Flux<ObjectRecord<String, String>> messages = receiver.receive(StreamOffset.fromStart("my-stream"));
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.then(() -> reactiveRedisTemplate.opsForStream().add(ObjectRecord.create("my-stream", "foobar")).subscribe())
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-stream");
|
||||
assertThat(it.getValue()).isEqualTo("foobar");
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveObjectHashRecords() {
|
||||
|
||||
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"));
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.then(() -> reactiveRedisTemplate.opsForStream()
|
||||
.add(ObjectRecord.create("my-logins", new LoginEvent("Walter", "White"))).subscribe())
|
||||
.consumeNextWith(it -> {
|
||||
|
||||
assertThat(it.getStream()).isEqualTo("my-logins");
|
||||
assertThat(it.getValue()).isEqualTo(new LoginEvent("Walter", "White"));
|
||||
}) //
|
||||
.thenCancel() //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
@@ -114,9 +173,10 @@ public class StreamReceiverIntegrationTests {
|
||||
|
||||
// XADD/XREAD highly timing-dependent as this tests require a poll subscription to receive messages using $ offset.
|
||||
|
||||
StreamReceiverOptions<String, String, String> options = StreamReceiverOptions.builder()
|
||||
StreamReceiverOptions<String, MapRecord<String, String, String>> options = StreamReceiverOptions.builder()
|
||||
.pollTimeout(Duration.ofSeconds(4)).build();
|
||||
StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory, options);
|
||||
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory,
|
||||
options);
|
||||
|
||||
Flux<MapRecord<String, String, String>> messages = receiver
|
||||
.receive(StreamOffset.create("my-stream", ReadOffset.latest()));
|
||||
@@ -126,18 +186,20 @@ public class StreamReceiverIntegrationTests {
|
||||
.then(() -> {
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"));
|
||||
reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value1"))
|
||||
.subscribe();
|
||||
} catch (InterruptedException e) {}
|
||||
}) //
|
||||
.expectNextCount(1) //
|
||||
.then(() -> {
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2"));
|
||||
reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value2")).subscribe();
|
||||
}) //
|
||||
.thenRequest(1) //
|
||||
.then(() -> {
|
||||
try {
|
||||
Thread.sleep(500);
|
||||
redisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3"));
|
||||
reactiveRedisTemplate.opsForStream().add("my-stream", Collections.singletonMap("key", "value3"))
|
||||
.subscribe();
|
||||
} catch (InterruptedException e) {}
|
||||
}).consumeNextWith(it -> {
|
||||
|
||||
@@ -151,7 +213,7 @@ public class StreamReceiverIntegrationTests {
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldReceiveAsConsumerGroupMessages() {
|
||||
|
||||
StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory);
|
||||
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory);
|
||||
|
||||
Flux<MapRecord<String, String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
|
||||
@@ -180,10 +242,11 @@ public class StreamReceiverIntegrationTests {
|
||||
@Test // DATAREDIS-864
|
||||
public void shouldStopReceivingOnError() {
|
||||
|
||||
StreamReceiverOptions<String, String, String> options = StreamReceiverOptions.builder()
|
||||
StreamReceiverOptions<String, MapRecord<String, String, String>> options = StreamReceiverOptions.builder()
|
||||
.pollTimeout(Duration.ofMillis(100)).build();
|
||||
|
||||
StreamReceiver<String, String, String> receiver = StreamReceiver.create(connectionFactory, options);
|
||||
StreamReceiver<String, MapRecord<String, String, String>> receiver = StreamReceiver.create(connectionFactory,
|
||||
options);
|
||||
|
||||
Flux<MapRecord<String, String, String>> messages = receiver.receive(Consumer.from("my-group", "my-consumer-id"),
|
||||
StreamOffset.create("my-stream", ReadOffset.lastConsumed()));
|
||||
@@ -194,8 +257,14 @@ public class StreamReceiverIntegrationTests {
|
||||
|
||||
messages.as(StepVerifier::create) //
|
||||
.expectNextCount(1) //
|
||||
.then(() -> redisTemplate.delete("my-stream")) //
|
||||
.then(() -> reactiveRedisTemplate.delete("my-stream").subscribe()) //
|
||||
.expectError(RedisSystemException.class) //
|
||||
.verify(Duration.ofSeconds(5));
|
||||
}
|
||||
|
||||
@Data
|
||||
@AllArgsConstructor
|
||||
static class LoginEvent {
|
||||
String firstname, lastname;
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user