From 8c4052329f8195c51ccdfea3846a4af1cd76a934 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Mon, 4 Jul 2022 15:06:32 +0200 Subject: [PATCH] Adopt to Reactor 2022.0.0-M4 changes. Closes #2354 --- .../LettuceConnectionFactoryTests.java | 2 + .../LettuceConnectionFactoryUnitTests.java | 16 ++++-- .../LettuceReactiveSubscriptionUnitTests.java | 40 +++++++-------- .../LettuceSentinelIntegrationTests.java | 1 + ...edisMessageListenerContainerUnitTests.java | 51 ++++++++++--------- .../extension/LettuceTestClientResources.java | 15 +++++- 6 files changed, 76 insertions(+), 49 deletions(-) diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java index 1be340af1..8cf22ca2b 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java @@ -33,6 +33,7 @@ import java.util.function.Consumer; import org.apache.commons.pool2.impl.GenericObjectPoolConfig; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.junit.jupiter.api.extension.ExtendWith; @@ -315,6 +316,7 @@ class LettuceConnectionFactoryTests { } @Test // DATAREDIS-462 + @Disabled("Until Lettuce upgrades to Sinks") void factoryWorksWithoutClientResources() { LettuceConnectionFactory factory = new LettuceConnectionFactory(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java index 3571f7d50..60f3ddfa5 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java @@ -44,6 +44,7 @@ import java.util.concurrent.CompletableFuture; import org.junit.jupiter.api.AfterEach; import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Disabled; import org.junit.jupiter.api.Test; import org.mockito.ArgumentMatchers; @@ -256,6 +257,7 @@ class LettuceConnectionFactoryUnitTests { } @Test // DATAREDIS-462 + @Disabled("Until Lettuce supports Sinks") void clusterClientShouldInitializeWithoutClientResources() { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig); @@ -470,6 +472,7 @@ class LettuceConnectionFactoryUnitTests { void socketShouldBeSetOnStandaloneClient() { LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(new RedisSocketConfiguration()); + connectionFactory.setClientResources(getSharedClientResources()); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -733,6 +736,7 @@ class LettuceConnectionFactoryUnitTests { connectionFactory.setShutdownTimeout(0); connectionFactory.setTimeout(2000); connectionFactory.setShareNativeConnection(false); + connectionFactory.setClientResources(getSharedClientResources()); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -745,8 +749,9 @@ class LettuceConnectionFactoryUnitTests { @Test // DATAREDIS-676 void timeoutSetOnClientConfigShouldBePassedOnToClusterConnection() { - LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, LettuceClientConfiguration - .builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO).build()); + LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, + LettuceClientConfiguration.builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO) + .clientResources(getSharedClientResources()).build()); connectionFactory.setShareNativeConnection(false); connectionFactory.afterPropertiesSet(); @@ -901,6 +906,7 @@ class LettuceConnectionFactoryUnitTests { } }; + connectionFactory.setClientResources(getSharedClientResources()); connectionFactory.afterPropertiesSet(); connectionFactory.destroy(); @@ -934,7 +940,7 @@ class LettuceConnectionFactoryUnitTests { clusterConfiguration.clusterNode("localhost", 1234).setMaxRedirects(42); LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfiguration, - LettuceClientConfiguration.defaultConfiguration()); + LettuceClientConfiguration.builder().clientResources(getSharedClientResources()).build()); connectionFactory.afterPropertiesSet(); ConnectionFactoryTracker.add(connectionFactory); @@ -954,7 +960,8 @@ class LettuceConnectionFactoryUnitTests { clusterConfiguration.clusterNode("localhost", 1234).setMaxRedirects(42); LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder() - .clientOptions(ClusterClientOptions.builder().validateClusterNodeMembership(false).build()).build(); + .clientOptions(ClusterClientOptions.builder().validateClusterNodeMembership(false).build()) + .clientResources(getSharedClientResources()).build(); LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfiguration, clientConfiguration); @@ -984,6 +991,7 @@ class LettuceConnectionFactoryUnitTests { return connectionProviderMock; } }; + connectionFactory.setClientResources(getSharedClientResources()); connectionFactory.afterPropertiesSet(); LettuceReactiveRedisConnection reactiveConnection = connectionFactory.getReactiveConnection(); diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java index 76f89afcf..75af51d28 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java @@ -24,9 +24,9 @@ import io.lettuce.core.RedisConnectionException; import io.lettuce.core.pubsub.StatefulRedisPubSubConnection; import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands; import reactor.core.Disposable; -import reactor.core.publisher.DirectProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; +import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; import java.nio.ByteBuffer; @@ -136,15 +136,15 @@ class LettuceReactiveSubscriptionUnitTests { when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); - DirectProcessor> emitter = DirectProcessor - .create(); - when(commandsMock.observeChannels()).thenReturn(emitter); + Sinks.Many> sink = Sinks.many().unicast() + .onBackpressureBuffer(); + when(commandsMock.observeChannels()).thenReturn(sink.asFlux()); when(commandsMock.observePatterns()).thenReturn(Flux.empty()); subscription.receive().as(StepVerifier::create).then(() -> { - emitter.onNext(createChannelMessage("other", "body")); - emitter.onNext(createChannelMessage("foo", "body")); + sink.tryEmitNext(createChannelMessage("other", "body")); + sink.tryEmitNext(createChannelMessage("foo", "body")); }).assertNext(msg -> { assertThat(msg.getChannel()).isEqualTo(getByteBuffer("foo")); }).thenCancel().verify(); @@ -156,15 +156,15 @@ class LettuceReactiveSubscriptionUnitTests { when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*")).as(StepVerifier::create).verifyComplete(); - DirectProcessor> emitter = DirectProcessor - .create(); + Sinks.Many> sink = Sinks.many().unicast() + .onBackpressureBuffer(); when(commandsMock.observeChannels()).thenReturn(Flux.empty()); - when(commandsMock.observePatterns()).thenReturn(emitter); + when(commandsMock.observePatterns()).thenReturn(sink.asFlux()); subscription.receive().as(StepVerifier::create).then(() -> { - emitter.onNext(createPatternMessage("other*", "channel", "body")); - emitter.onNext(createPatternMessage("foo*", "foo", "body")); + sink.tryEmitNext(createPatternMessage("other*", "channel", "body")); + sink.tryEmitNext(createPatternMessage("foo*", "foo", "body")); }).assertNext(msg -> { assertThat(((PatternMessage) msg).getPattern()).isEqualTo(getByteBuffer("foo*")); @@ -178,14 +178,14 @@ class LettuceReactiveSubscriptionUnitTests { when(commandsMock.subscribe(any())).thenReturn(Mono.empty()); subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete(); - DirectProcessor> emitter = DirectProcessor - .create(); - when(commandsMock.observeChannels()).thenReturn(emitter); + Sinks.Many> sink = Sinks.many().unicast() + .onBackpressureBuffer(); + when(commandsMock.observeChannels()).thenReturn(sink.asFlux()); when(commandsMock.observePatterns()).thenReturn(Flux.empty()); subscription.receive().as(StepVerifier::create).then(() -> { - emitter.onError(new RedisConnectionException("foo")); + sink.tryEmitError(new RedisConnectionException("foo")); }).expectError(RedisSystemException.class).verify(); } @@ -209,22 +209,22 @@ class LettuceReactiveSubscriptionUnitTests { @Test // DATAREDIS-612 void cancelledSubscriptionShouldUnregisterDownstream() { - DirectProcessor> emitter = DirectProcessor - .create(); + Sinks.Many> sink = Sinks.many().unicast() + .onBackpressureBuffer(); when(commandsMock.psubscribe(any())).thenReturn(Mono.empty()); subscription.pSubscribe(getByteBuffer("foo*")).as(StepVerifier::create).verifyComplete(); when(commandsMock.observeChannels()).thenReturn(Flux.never()); - when(commandsMock.observePatterns()).thenReturn(emitter); + when(commandsMock.observePatterns()).thenReturn(sink.asFlux()); Flux> receive = subscription.receive(); Disposable subscribe = receive.subscribe(); - assertThat(emitter.downstreamCount()).isEqualTo(1); + assertThat(sink.currentSubscriberCount()).isEqualTo(1); subscribe.dispose(); - assertThat(emitter.downstreamCount()).isEqualTo(0); + assertThat(sink.currentSubscriberCount()).isEqualTo(0); } private static io.lettuce.core.pubsub.api.reactive.ChannelMessage createChannelMessage( diff --git a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java index 82c54fedd..7bfb20740 100644 --- a/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java +++ b/src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java @@ -171,6 +171,7 @@ public class LettuceSentinelIntegrationTests extends AbstractConnectionIntegrati } @Test // DATAREDIS-462 + @Disabled("Until Lettuce has moved to Sinks API") void factoryWorksWithoutClientResources() { LettuceConnectionFactory factory = new LettuceConnectionFactory(SENTINEL_CONFIG); diff --git a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java index 91ccd568f..d344000a2 100644 --- a/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java +++ b/src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java @@ -20,10 +20,9 @@ import static org.mockito.Mockito.*; import static org.springframework.data.redis.util.ByteUtils.*; import reactor.core.Disposable; -import reactor.core.publisher.DirectProcessor; import reactor.core.publisher.Flux; import reactor.core.publisher.Mono; -import reactor.core.publisher.MonoProcessor; +import reactor.core.publisher.Sinks; import reactor.test.StepVerifier; import java.nio.ByteBuffer; @@ -92,8 +91,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { container = createContainer(); container.receive(PatternTopic.of("foo*"), PatternTopic.of("bar*")).as(StepVerifier::create).thenRequest(1) - .thenAwait() - .thenCancel().verify(); + .thenAwait().thenCancel().verify(); verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*")); } @@ -124,15 +122,15 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 void shouldEmitChannelMessage() { - DirectProcessor> processor = DirectProcessor.create(); + Sinks.Many> sink = Sinks.many().unicast().onBackpressureBuffer(); - when(subscriptionMock.receive()).thenReturn(processor); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(ChannelTopic.of("foo")); messageStream.as(StepVerifier::create).then(() -> { - processor.onNext(createChannelMessage("foo", "message")); + sink.tryEmitNext(createChannelMessage("foo", "message")); }).assertNext(msg -> { assertThat(msg.getChannel()).isEqualTo("foo"); @@ -143,15 +141,15 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 void shouldEmitPatternMessage() { - DirectProcessor> processor = DirectProcessor.create(); + Sinks.Many> sink = Sinks.many().unicast().onBackpressureBuffer(); - when(subscriptionMock.receive()).thenReturn(processor); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(PatternTopic.of("foo*")); messageStream.as(StepVerifier::create).then(() -> { - processor.onNext(createPatternMessage("foo*", "foo", "message")); + sink.tryEmitNext(createPatternMessage("foo*", "foo", "message")); }).assertNext(msg -> { assertThat(msg.getPattern()).isEqualTo("foo*"); @@ -163,12 +161,14 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 void shouldRegisterSubscription() { - MonoProcessor subscribeMono = MonoProcessor.create(); + Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(); + + Sinks.One subscribeMono = Sinks.one(); reset(subscriptionMock); - when(subscriptionMock.subscribe(any())).thenReturn(subscribeMono); + when(subscriptionMock.subscribe(any())).thenReturn(subscribeMono.asMono()); when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty()); - when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(ChannelTopic.of("foo*")); @@ -176,7 +176,7 @@ class ReactiveRedisMessageListenerContainerUnitTests { Disposable subscription = messageStream.subscribe(); assertThat(container.getActiveSubscriptions()).isEmpty(); - subscribeMono.onComplete(); + subscribeMono.tryEmitEmpty(); assertThat(container.getActiveSubscriptions()).isNotEmpty(); subscription.dispose(); assertThat(container.getActiveSubscriptions()).isEmpty(); @@ -185,10 +185,12 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612, GH-1622 void shouldRegisterSubscriptionMultipleSubscribers() { + Sinks.Many> sink = Sinks.many().multicast().onBackpressureBuffer(); + reset(subscriptionMock); when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty()); when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty()); - when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(new ChannelTopic("foo*")); @@ -210,7 +212,8 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612, GH-1622 void shouldUnsubscribeOnCancel() { - when(subscriptionMock.receive()).thenReturn(DirectProcessor.create()); + Sinks.Many> sink = Sinks.many().unicast().onBackpressureBuffer(); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(PatternTopic.of("foo*")); @@ -227,12 +230,12 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 void shouldTerminateSubscriptionsOnShutdown() { - DirectProcessor> processor = DirectProcessor.create(); + Sinks.Many> sink = Sinks.many().unicast().onBackpressureBuffer(); - when(subscriptionMock.receive()).thenReturn(processor); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); when(subscriptionMock.cancel()).thenReturn(Mono.defer(() -> { - processor.onError(new CancellationException()); + sink.tryEmitError(new CancellationException()); return Mono.empty(); })); container = createContainer(); @@ -247,19 +250,19 @@ class ReactiveRedisMessageListenerContainerUnitTests { @Test // DATAREDIS-612 void shouldCleanupDownstream() { - DirectProcessor> processor = DirectProcessor.create(); + Sinks.Many> sink = Sinks.many().unicast().onBackpressureBuffer(); - when(subscriptionMock.receive()).thenReturn(processor); + when(subscriptionMock.receive()).thenReturn(sink.asFlux()); container = createContainer(); Flux> messageStream = container.receive(PatternTopic.of("foo*")); messageStream.as(StepVerifier::create).then(() -> { - assertThat(processor.hasDownstreams()).isTrue(); - processor.onNext(createPatternMessage("foo*", "foo", "message")); + assertThat(sink.currentSubscriberCount()).isGreaterThan(0); + sink.tryEmitNext(createPatternMessage("foo*", "foo", "message")); }).expectNextCount(1).thenCancel().verify(); - assertThat(processor.hasDownstreams()).isFalse(); + assertThat(sink.currentSubscriberCount()).isEqualTo(0); } private ReactiveRedisMessageListenerContainer createContainer() { diff --git a/src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java b/src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java index 9ca7c832a..274ffa021 100644 --- a/src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java +++ b/src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java @@ -15,8 +15,11 @@ */ package org.springframework.data.redis.test.extension; +import io.lettuce.core.event.Event; +import io.lettuce.core.event.EventBus; import io.lettuce.core.resource.ClientResources; import io.lettuce.core.resource.DefaultClientResources; +import reactor.core.publisher.Flux; import java.util.concurrent.TimeUnit; @@ -33,7 +36,17 @@ public class LettuceTestClientResources { static { - SHARED_CLIENT_RESOURCES = DefaultClientResources.builder() + SHARED_CLIENT_RESOURCES = DefaultClientResources.builder().eventBus(new EventBus() { + @Override + public Flux get() { + return Flux.empty(); + } + + @Override + public void publish(Event event) { + + } + }) .build(); ShutdownQueue.INSTANCE.register(() -> SHARED_CLIENT_RESOURCES.shutdown(0, 0, TimeUnit.MILLISECONDS)); }