Spring/spring-data-redis
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

Adopt to Reactor 2022.0.0-M4 changes.

Browse Source 
Closes #2354
This commit is contained in:
Mark Paluch
2022-07-04 15:06:32 +02:00
parent 105964394e
commit 8c4052329f
6 changed files with 76 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java
View File

@@ -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();

16
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java
View File

@@ -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();

40
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java
View File

@@ -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<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
.create();
when(commandsMock.observeChannels()).thenReturn(emitter);
Sinks.Many<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> 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<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
.create();
Sinks.Many<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> 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<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
.create();
when(commandsMock.observeChannels()).thenReturn(emitter);
Sinks.Many<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> 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<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
.create();
Sinks.Many<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> 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<Message<ByteBuffer, ByteBuffer>> 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<ByteBuffer, ByteBuffer> createChannelMessage(

1
src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java
View File

@@ -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);

51
src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java
View File

@@ -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<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
when(subscriptionMock.receive()).thenReturn(processor);
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
container = createContainer();
Flux<Message<String, String>> 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<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
when(subscriptionMock.receive()).thenReturn(processor);
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
container = createContainer();
Flux<PatternMessage<String, String, String>> 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<Void> subscribeMono = MonoProcessor.create();
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().multicast().onBackpressureBuffer();
Sinks.One<Void> 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<Message<String, String>> 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<Message<ByteBuffer, ByteBuffer>> 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<Message<String, String>> 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<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
container = createContainer();
Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
@@ -227,12 +230,12 @@ class ReactiveRedisMessageListenerContainerUnitTests {
@Test // DATAREDIS-612
void shouldTerminateSubscriptionsOnShutdown() {
DirectProcessor<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
Sinks.Many<Message<ByteBuffer, ByteBuffer>> 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<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
when(subscriptionMock.receive()).thenReturn(processor);
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
container = createContainer();
Flux<PatternMessage<String, String, String>> 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() {

15
src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java
View File

@@ -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<Event> get() {
return Flux.empty();
}
@Override
public void publish(Event event) {
}
})
.build();
ShutdownQueue.INSTANCE.register(() -> SHARED_CLIENT_RESOURCES.shutdown(0, 0, TimeUnit.MILLISECONDS));
}