diff --git a/src/main/asciidoc/reference/redis-streams.adoc b/src/main/asciidoc/reference/redis-streams.adoc index 0574d7024..f6b6b32be 100644 --- a/src/main/asciidoc/reference/redis-streams.adoc +++ b/src/main/asciidoc/reference/redis-streams.adoc @@ -25,7 +25,7 @@ To send a record, you can use, as with the other operations, either the low-leve [source,java] ---- -// append message through connection +// append message through connection RedisConnection con = … byte[] stream = … ByteRecord record = StreamRecords.rawBytes(…).withStreamKey(stream); @@ -114,18 +114,18 @@ message -> { ---- Once you’ve implemented your `StreamListener`, it’s time to create a message listener container and register a subscription: - + [source,java] ---- RedisConnectionFactory connectionFactory = … StreamListener> streamListener = … - + StreamMessageListenerContainerOptions> containerOptions = StreamMessageListenerContainerOptions .builder().pollTimeout(Duration.ofMillis(100)).build(); - + StreamMessageListenerContainer> container = StreamMessageListenerContainer.create(connectionFactory, containerOptions); - + Subscription subscription = container.receive(StreamOffset.fromStart("my-stream"), streamListener); ---- @@ -133,7 +133,7 @@ Please refer to the Javadoc of the various message listener containers for a ful ==== Reactive `StreamReceiver` -Reactive consumption of streaming data sources typically happens through a `Flux` of events or messages. The reactive receiver implementation is provided with `StreamReceiver` and its overloaded `receive(…)` messages. The reactive approach requires fewer infrastructure resources such as threads in comparison to `StreamMessageListenerContainer` as it is leveraging threading resources provided by the driver. The receiving stream is a demand-driven publisher of ``StreamMessage``: +Reactive consumption of streaming data sources typically happens through a `Flux` of events or messages. The reactive receiver implementation is provided with `StreamReceiver` and its overloaded `receive(…)` messages. The reactive approach requires fewer infrastructure resources such as threads in comparison to `StreamMessageListenerContainer` as it is leveraging threading resources provided by the driver. The receiving stream is a demand-driven publisher of ``StreamMessage``: [source,java] ---- @@ -155,7 +155,7 @@ ReactiveRedisConnectionFactory connectionFactory = … StreamReceiverOptions> options = StreamReceiverOptions.builder().pollTimeout(Duration.ofMillis(100)) .build(); StreamReceiver> receiver = StreamReceiver.create(connectionFactory, options); - + Flux> messages = receiver.receive(StreamOffset.fromStart("my-stream")); ---- @@ -296,34 +296,35 @@ redisTemplate() [NOTE] ==== -A `StreamMessageListenerContainer` may not be aware of any `@TypeAlias` used on domain types as those need to be resolved via a `MappingContext`. So please make sure to pre initialize the `RedisMappingContext` with the `initialEntitySet`. +A `StreamMessageListenerContainer` may not be aware of any `@TypeAlias` used on domain types as those need to be resolved through a `MappingContext`. +Make sure to initialize `RedisMappingContext` with a `initialEntitySet`. [source,java] ---- @Bean RedisMappingContext redisMappingContext() { - RedisMappingContext ctx = new RedisMappingContext(); - ctx.setInitialEntitySet(Collections.singleton(Person.class)); - return ctx; + RedisMappingContext ctx = new RedisMappingContext(); + ctx.setInitialEntitySet(Collections.singleton(Person.class)); + return ctx; } @Bean RedisConverter redisConverter(RedisMappingContext mappingContext) { - return new MappingRedisConverter(mappingContext, null, null); + return new MappingRedisConverter(mappingContext); } @Bean ObjectHashMapper hashMapper(RedisConverter converter) { - return new ObjectHashMapper(converter); + return new ObjectHashMapper(converter); } @Bean StreamMessageListenerContainer streamMessageListenerContainer(RedisConnectionFactory connectionFactory, ObjectHashMapper hashMapper) { - StreamMessageListenerContainerOptions> options = StreamMessageListenerContainerOptions.builder() - .objectMapper(hashMapper) - .build(); + StreamMessageListenerContainerOptions> options = StreamMessageListenerContainerOptions.builder() + .objectMapper(hashMapper) + .build(); - return StreamMessageListenerContainer.create(connectionFactory, options); + return StreamMessageListenerContainer.create(connectionFactory, options); } ---- ==== diff --git a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java index 6329fe6e2..fc8a8a764 100644 --- a/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java +++ b/src/main/java/org/springframework/data/redis/core/convert/MappingRedisConverter.java @@ -129,8 +129,9 @@ public class MappingRedisConverter implements RedisConverter, InitializingBean { * Creates new {@link MappingRedisConverter}. * * @param context can be {@literal null}. + * @since 2.4 */ - MappingRedisConverter(RedisMappingContext context) { + public MappingRedisConverter(RedisMappingContext context) { this(context, null, null, null); } diff --git a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java index bed73efe8..58d8986c8 100644 --- a/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java +++ b/src/main/java/org/springframework/data/redis/hash/ObjectHashMapper.java @@ -80,6 +80,19 @@ public class ObjectHashMapper implements HashMapper { this(new RedisCustomConversions()); } + /** + * Creates a new {@link ObjectHashMapper} using the given {@link RedisConverter} for conversion. + * + * @param converter must not be {@literal null}. + * @throws IllegalArgumentException if the given {@literal converter} is {@literal null}. + * @since 2.4 + */ + public ObjectHashMapper(RedisConverter converter) { + + Assert.notNull(converter, "Converter must not be null!"); + this.converter = converter; + } + /** * Creates new {@link ObjectHashMapper}. * @@ -107,19 +120,6 @@ public class ObjectHashMapper implements HashMapper { converter = mappingConverter; } - /** - * Creates a new {@link ObjectHashMapper} using the given {@link RedisConverter} for conversion. - * - * @param converter must not be {@literal null}. - * @throws IllegalArgumentException if the given {@literal converter} is {@literal null}. - * @since 2.4 - */ - public ObjectHashMapper(RedisConverter converter) { - - Assert.notNull(converter, "Converter must not be null!"); - this.converter = converter; - } - /* * (non-Javadoc) * @see org.springframework.data.redis.hash.HashMapper#toHash(java.lang.Object) diff --git a/src/test/java/org/springframework/data/redis/mapping/ObjectHashMapperTests.java b/src/test/java/org/springframework/data/redis/mapping/ObjectHashMapperTests.java index c09b39952..1d46b9ccb 100644 --- a/src/test/java/org/springframework/data/redis/mapping/ObjectHashMapperTests.java +++ b/src/test/java/org/springframework/data/redis/mapping/ObjectHashMapperTests.java @@ -29,6 +29,8 @@ import org.springframework.data.redis.core.mapping.RedisMappingContext; import org.springframework.data.redis.hash.ObjectHashMapper; /** + * Unit tests for {@link ObjectHashMapper}. + * * @author Christoph Strobl * @author Mark Paluch */