diff --git a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java index b613996ec6..d60f76c5b3 100644 --- a/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java +++ b/spring-integration-r2dbc/src/main/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandler.java @@ -112,7 +112,7 @@ public class R2dbcMessageHandler extends AbstractReactiveMessageHandler { @Override public String getComponentType() { - return "r2dbc:reactive-outbound-channel-adapter"; + return "r2dbc:outbound-channel-adapter"; } @Override diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java index a0e51636f1..9c7207a1d0 100644 --- a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/inbound/R2dbcMessageSourceTests.java @@ -89,6 +89,12 @@ public class R2dbcMessageSourceTests { .verifyComplete()); } + @Test + public void validateComponentType() { + assertThat(this.defaultR2dbcMessageSource.getComponentType()) + .isEqualTo("r2dbc:inbound-channel-adapter"); + } + @Test public void validateSuccessfulQueryWithoutSettingExpectedElement() { this.entityTemplate.insert(new Person("Bob", 35)) diff --git a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java index 8c499a6183..c8cf5417fc 100644 --- a/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java +++ b/spring-integration-r2dbc/src/test/java/org/springframework/integration/r2dbc/outbound/R2dbcMessageHandlerTests.java @@ -16,6 +16,7 @@ package org.springframework.integration.r2dbc.outbound; +import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; import java.util.Arrays; @@ -115,6 +116,11 @@ public class R2dbcMessageHandlerTests { } + @Test + public void validateComponentType() { + assertThat(this.r2dbcMessageHandler.getComponentType()).isEqualTo("r2dbc:outbound-channel-adapter"); + } + @Test public void validateMessageHandlingWithDefaultUpdateCollection() { Message message = MessageBuilder.withPayload(createPerson("Bob", 35)).build(); diff --git a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java index 6ab5a15e02..6c27700650 100644 --- a/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java +++ b/spring-integration-redis/src/main/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducer.java @@ -28,6 +28,7 @@ import org.springframework.data.redis.core.ReactiveStreamOperations; import org.springframework.data.redis.serializer.RedisSerializationContext; import org.springframework.data.redis.stream.StreamReceiver; import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.acks.SimpleAcknowledgment; import org.springframework.integration.endpoint.MessageProducerSupport; import org.springframework.integration.redis.support.RedisHeaders; import org.springframework.integration.support.AbstractIntegrationMessageBuilder; @@ -48,6 +49,7 @@ import reactor.core.publisher.Mono; * * @author Attoumane Ahamadi * @author Artem Bilan + * @author Rohan Mukesh * * @since 5.4 */ @@ -217,10 +219,12 @@ public class ReactiveRedisStreamMessageProducer extends MessageProducerSupport { .setHeader(RedisHeaders.STREAM_MESSAGE_ID, event.getId()) .setHeader(RedisHeaders.CONSUMER_GROUP, this.consumerGroup) .setHeader(RedisHeaders.CONSUMER, this.consumerName); - if (!this.autoAck) { + if (!this.autoAck && this.consumerGroup != null) { builder.setHeader(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK, - this.reactiveStreamOperations.acknowledge(this.consumerGroup, event) - .subscribe()); + (SimpleAcknowledgment) () -> + this.reactiveStreamOperations + .acknowledge(this.consumerGroup, event) + .subscribe()); } return builder.build(); }); diff --git a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java index 950b2be6f6..de74e16f4c 100644 --- a/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java +++ b/spring-integration-redis/src/test/java/org/springframework/integration/redis/inbound/ReactiveRedisStreamMessageProducerTests.java @@ -19,6 +19,7 @@ package org.springframework.integration.redis.inbound; import static org.assertj.core.api.Assertions.assertThat; import java.time.Duration; +import java.util.concurrent.atomic.AtomicReference; import org.junit.After; import org.junit.Before; @@ -28,10 +29,15 @@ import org.junit.runner.RunWith; import org.springframework.beans.factory.annotation.Autowired; import org.springframework.context.annotation.Bean; import org.springframework.context.annotation.Configuration; +import org.springframework.data.redis.connection.stream.PendingMessagesSummary; import org.springframework.data.redis.connection.stream.ReadOffset; +import org.springframework.data.redis.connection.stream.StreamInfo; import org.springframework.data.redis.core.ReactiveRedisTemplate; import org.springframework.data.redis.serializer.RedisSerializationContext; import org.springframework.data.redis.stream.StreamReceiver; +import org.springframework.integration.IntegrationMessageHeaderAccessor; +import org.springframework.integration.StaticMessageHeaderAccessor; +import org.springframework.integration.acks.SimpleAcknowledgment; import org.springframework.integration.channel.FluxMessageChannel; import org.springframework.integration.handler.ReactiveMessageHandlerAdapter; import org.springframework.integration.redis.outbound.ReactiveRedisStreamMessageHandler; @@ -46,11 +52,13 @@ import org.springframework.test.annotation.DirtiesContext; import org.springframework.test.context.junit4.SpringRunner; import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; import reactor.test.StepVerifier; /** * @author Attoumane Ahamadi * @author Artem Bilan + * @author Rohan Mukesh * * @since 5.4 */ @@ -76,6 +84,16 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests @Before public void delKey() { + this.template.hasKey(STREAM_KEY) + .filter(Boolean::booleanValue) + .flatMapMany(b -> + this.template.opsForStream() + .groups(STREAM_KEY) + .map(StreamInfo.XInfoGroup::groupName) + .flatMap(groupName -> + this.template.opsForStream() + .destroyGroup(STREAM_KEY, groupName))) + .blockLast(); this.template.delete(STREAM_KEY).block(); } @@ -90,10 +108,11 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests this.redisStreamMessageProducer.setCreateConsumerGroup(true); this.redisStreamMessageProducer.setConsumerName(CONSUMER); this.redisStreamMessageProducer.afterPropertiesSet(); - this.redisStreamMessageProducer.start(); Flux.from(this.fluxMessageChannel).subscribe(); + this.redisStreamMessageProducer.start(); + this.template.opsForStream() .groups(STREAM_KEY) .next() @@ -147,19 +166,81 @@ public class ReactiveRedisStreamMessageProducerTests extends RedisAvailableTests this.redisStreamMessageProducer.setCreateConsumerGroup(false); this.redisStreamMessageProducer.setConsumerName(CONSUMER); + this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest()); this.redisStreamMessageProducer.afterPropertiesSet(); this.redisStreamMessageProducer.start(); + StepVerifier stepVerifier = + Flux.from(this.fluxMessageChannel) + .as(StepVerifier::create) + .assertNext(message -> { + assertThat(message.getPayload()).isEqualTo(person); + assertThat(message.getHeaders()).containsKeys(RedisHeaders.CONSUMER_GROUP, RedisHeaders.CONSUMER); + }) + .thenCancel() + .verifyLater(); + this.messageHandler.handleMessage(new GenericMessage<>(person)); - Flux.from(this.fluxMessageChannel) + stepVerifier.verify(Duration.ofSeconds(10)); + } + + @Test + @RedisAvailable + public void testReadingPendingMessageWithNoAutoACK() { + Address address = new Address("Winterfell, Westeros"); + Person person = new Person(address, "John Snow"); + + this.template.opsForStream() + .createGroup(STREAM_KEY, this.redisStreamMessageProducer.getBeanName()) .as(StepVerifier::create) - .assertNext(message -> { - assertThat(message.getPayload()).isEqualTo(person); - assertThat(message.getHeaders()).containsKeys(RedisHeaders.CONSUMER_GROUP, RedisHeaders.CONSUMER); - }) + .assertNext(message -> assertThat(message).isEqualTo("OK")) .thenCancel() .verify(Duration.ofSeconds(10)); + + this.redisStreamMessageProducer.setCreateConsumerGroup(false); + this.redisStreamMessageProducer.setAutoAck(false); + this.redisStreamMessageProducer.setConsumerName(CONSUMER); + this.redisStreamMessageProducer.setReadOffset(ReadOffset.latest()); + this.redisStreamMessageProducer.afterPropertiesSet(); + this.redisStreamMessageProducer.start(); + + AtomicReference acknowledgmentReference = new AtomicReference<>(); + + StepVerifier stepVerifier = + Flux.from(this.fluxMessageChannel) + .as(StepVerifier::create) + .assertNext(message -> { + assertThat(message.getPayload()).isEqualTo(person); + assertThat(message.getHeaders()) + .containsKeys(IntegrationMessageHeaderAccessor.ACKNOWLEDGMENT_CALLBACK); + acknowledgmentReference.set(StaticMessageHeaderAccessor.getAcknowledgment(message)); + }) + .thenCancel() + .verifyLater(); + + this.messageHandler.handleMessage(new GenericMessage<>(person)); + + stepVerifier.verify(Duration.ofSeconds(10)); + + Mono pending = + template.opsForStream() + .pending(STREAM_KEY, this.redisStreamMessageProducer.getBeanName()); + + StepVerifier.create(pending) + .assertNext(pendingMessagesSummary -> + assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(1)) + .verifyComplete(); + + acknowledgmentReference.get().acknowledge(); + + Mono pendingZeroMessage = template.opsForStream().pending(STREAM_KEY, + this.redisStreamMessageProducer.getBeanName()); + + StepVerifier.create(pendingZeroMessage) + .assertNext(pendingMessagesSummary -> + assertThat(pendingMessagesSummary.getTotalPendingMessages()).isEqualTo(0)) + .verifyComplete(); } @Configuration