Skip to content

Commit 8c40523

Browse files
mp911demp911de
committed
Adopt to Reactor 2022.0.0-M4 changes.
Closes #2354
1 parent 1059643 commit 8c40523

6 files changed

+76
-49
lines changed

‎src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryTests.java

+2
Original file line numberDiff line numberDiff line change
@@ -33,6 +33,7 @@
3333
import org.apache.commons.pool2.impl.GenericObjectPoolConfig;
3434
import org.junit.jupiter.api.AfterEach;
3535
import org.junit.jupiter.api.BeforeEach;
36+
import org.junit.jupiter.api.Disabled;
3637
import org.junit.jupiter.api.Test;
3738
import org.junit.jupiter.api.extension.ExtendWith;
3839

@@ -315,6 +316,7 @@ void dbIndexShouldBePropagatedCorrectly() {
315316
}
316317

317318
@Test // DATAREDIS-462
319+
@Disabled("Until Lettuce upgrades to Sinks")
318320
void factoryWorksWithoutClientResources() {
319321

320322
LettuceConnectionFactory factory = new LettuceConnectionFactory();

‎src/test/java/org/springframework/data/redis/connection/lettuce/LettuceConnectionFactoryUnitTests.java

+12-4
Original file line numberDiff line numberDiff line change
@@ -44,6 +44,7 @@
4444

4545
import org.junit.jupiter.api.AfterEach;
4646
import org.junit.jupiter.api.BeforeEach;
47+
import org.junit.jupiter.api.Disabled;
4748
import org.junit.jupiter.api.Test;
4849
import org.mockito.ArgumentMatchers;
4950

@@ -256,6 +257,7 @@ void sentinelPasswordShouldNotLeakIntoDataNodeClient() {
256257
}
257258

258259
@Test // DATAREDIS-462
260+
@Disabled("Until Lettuce supports Sinks")
259261
void clusterClientShouldInitializeWithoutClientResources() {
260262

261263
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig);
@@ -470,6 +472,7 @@ void verifyPeerTLSOptionShouldBeSetCorrectlyOnClusterClient() {
470472
void socketShouldBeSetOnStandaloneClient() {
471473

472474
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(new RedisSocketConfiguration());
475+
connectionFactory.setClientResources(getSharedClientResources());
473476
connectionFactory.afterPropertiesSet();
474477
ConnectionFactoryTracker.add(connectionFactory);
475478

@@ -733,6 +736,7 @@ void timeoutShouldBePassedOnToClusterConnection() {
733736
connectionFactory.setShutdownTimeout(0);
734737
connectionFactory.setTimeout(2000);
735738
connectionFactory.setShareNativeConnection(false);
739+
connectionFactory.setClientResources(getSharedClientResources());
736740
connectionFactory.afterPropertiesSet();
737741
ConnectionFactoryTracker.add(connectionFactory);
738742

@@ -745,8 +749,9 @@ void timeoutShouldBePassedOnToClusterConnection() {
745749
@Test // DATAREDIS-676
746750
void timeoutSetOnClientConfigShouldBePassedOnToClusterConnection() {
747751

748-
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig, LettuceClientConfiguration
749-
.builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO).build());
752+
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfig,
753+
LettuceClientConfiguration.builder().commandTimeout(Duration.ofSeconds(2)).shutdownTimeout(Duration.ZERO)
754+
.clientResources(getSharedClientResources()).build());
750755
connectionFactory.setShareNativeConnection(false);
751756

752757
connectionFactory.afterPropertiesSet();
@@ -901,6 +906,7 @@ protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClie
901906
}
902907
};
903908

909+
connectionFactory.setClientResources(getSharedClientResources());
904910
connectionFactory.afterPropertiesSet();
905911
connectionFactory.destroy();
906912

@@ -934,7 +940,7 @@ void maxRedirectsShouldBeSetOnClientOptions() {
934940
clusterConfiguration.clusterNode("localhost", 1234).setMaxRedirects(42);
935941

936942
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfiguration,
937-
LettuceClientConfiguration.defaultConfiguration());
943+
LettuceClientConfiguration.builder().clientResources(getSharedClientResources()).build());
938944
connectionFactory.afterPropertiesSet();
939945
ConnectionFactoryTracker.add(connectionFactory);
940946

@@ -954,7 +960,8 @@ void maxRedirectsShouldBeSetOnClusterClientOptions() {
954960
clusterConfiguration.clusterNode("localhost", 1234).setMaxRedirects(42);
955961

956962
LettuceClientConfiguration clientConfiguration = LettuceClientConfiguration.builder()
957-
.clientOptions(ClusterClientOptions.builder().validateClusterNodeMembership(false).build()).build();
963+
.clientOptions(ClusterClientOptions.builder().validateClusterNodeMembership(false).build())
964+
.clientResources(getSharedClientResources()).build();
958965

959966
LettuceConnectionFactory connectionFactory = new LettuceConnectionFactory(clusterConfiguration,
960967
clientConfiguration);
@@ -984,6 +991,7 @@ protected LettuceConnectionProvider doCreateConnectionProvider(AbstractRedisClie
984991
return connectionProviderMock;
985992
}
986993
};
994+
connectionFactory.setClientResources(getSharedClientResources());
987995
connectionFactory.afterPropertiesSet();
988996

989997
LettuceReactiveRedisConnection reactiveConnection = connectionFactory.getReactiveConnection();

‎src/test/java/org/springframework/data/redis/connection/lettuce/LettuceReactiveSubscriptionUnitTests.java

+20-20
Original file line numberDiff line numberDiff line change
@@ -24,9 +24,9 @@
2424
import io.lettuce.core.pubsub.StatefulRedisPubSubConnection;
2525
import io.lettuce.core.pubsub.api.reactive.RedisPubSubReactiveCommands;
2626
import reactor.core.Disposable;
27-
import reactor.core.publisher.DirectProcessor;
2827
import reactor.core.publisher.Flux;
2928
import reactor.core.publisher.Mono;
29+
import reactor.core.publisher.Sinks;
3030
import reactor.test.StepVerifier;
3131

3232
import java.nio.ByteBuffer;
@@ -136,15 +136,15 @@ void shouldEmitChannelMessage() {
136136
when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
137137
subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete();
138138

139-
DirectProcessor<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
140-
.create();
141-
when(commandsMock.observeChannels()).thenReturn(emitter);
139+
Sinks.Many<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast()
140+
.onBackpressureBuffer();
141+
when(commandsMock.observeChannels()).thenReturn(sink.asFlux());
142142
when(commandsMock.observePatterns()).thenReturn(Flux.empty());
143143

144144
subscription.receive().as(StepVerifier::create).then(() -> {
145145

146-
emitter.onNext(createChannelMessage("other", "body"));
147-
emitter.onNext(createChannelMessage("foo", "body"));
146+
sink.tryEmitNext(createChannelMessage("other", "body"));
147+
sink.tryEmitNext(createChannelMessage("foo", "body"));
148148
}).assertNext(msg -> {
149149
assertThat(msg.getChannel()).isEqualTo(getByteBuffer("foo"));
150150
}).thenCancel().verify();
@@ -156,15 +156,15 @@ void shouldEmitPatternMessage() {
156156
when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
157157
subscription.pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*")).as(StepVerifier::create).verifyComplete();
158158

159-
DirectProcessor<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
160-
.create();
159+
Sinks.Many<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast()
160+
.onBackpressureBuffer();
161161
when(commandsMock.observeChannels()).thenReturn(Flux.empty());
162-
when(commandsMock.observePatterns()).thenReturn(emitter);
162+
when(commandsMock.observePatterns()).thenReturn(sink.asFlux());
163163

164164
subscription.receive().as(StepVerifier::create).then(() -> {
165165

166-
emitter.onNext(createPatternMessage("other*", "channel", "body"));
167-
emitter.onNext(createPatternMessage("foo*", "foo", "body"));
166+
sink.tryEmitNext(createPatternMessage("other*", "channel", "body"));
167+
sink.tryEmitNext(createPatternMessage("foo*", "foo", "body"));
168168
}).assertNext(msg -> {
169169

170170
assertThat(((PatternMessage) msg).getPattern()).isEqualTo(getByteBuffer("foo*"));
@@ -178,14 +178,14 @@ void shouldEmitError() {
178178
when(commandsMock.subscribe(any())).thenReturn(Mono.empty());
179179
subscription.subscribe(getByteBuffer("foo"), getByteBuffer("bar")).as(StepVerifier::create).verifyComplete();
180180

181-
DirectProcessor<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
182-
.create();
183-
when(commandsMock.observeChannels()).thenReturn(emitter);
181+
Sinks.Many<io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast()
182+
.onBackpressureBuffer();
183+
when(commandsMock.observeChannels()).thenReturn(sink.asFlux());
184184
when(commandsMock.observePatterns()).thenReturn(Flux.empty());
185185

186186
subscription.receive().as(StepVerifier::create).then(() -> {
187187

188-
emitter.onError(new RedisConnectionException("foo"));
188+
sink.tryEmitError(new RedisConnectionException("foo"));
189189
}).expectError(RedisSystemException.class).verify();
190190
}
191191

@@ -209,22 +209,22 @@ void shouldTerminateActiveSubscriptions() {
209209
@Test // DATAREDIS-612
210210
void cancelledSubscriptionShouldUnregisterDownstream() {
211211

212-
DirectProcessor<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> emitter = DirectProcessor
213-
.create();
212+
Sinks.Many<io.lettuce.core.pubsub.api.reactive.PatternMessage<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast()
213+
.onBackpressureBuffer();
214214

215215
when(commandsMock.psubscribe(any())).thenReturn(Mono.empty());
216216
subscription.pSubscribe(getByteBuffer("foo*")).as(StepVerifier::create).verifyComplete();
217217

218218
when(commandsMock.observeChannels()).thenReturn(Flux.never());
219-
when(commandsMock.observePatterns()).thenReturn(emitter);
219+
when(commandsMock.observePatterns()).thenReturn(sink.asFlux());
220220

221221
Flux<Message<ByteBuffer, ByteBuffer>> receive = subscription.receive();
222222
Disposable subscribe = receive.subscribe();
223223

224-
assertThat(emitter.downstreamCount()).isEqualTo(1);
224+
assertThat(sink.currentSubscriberCount()).isEqualTo(1);
225225

226226
subscribe.dispose();
227-
assertThat(emitter.downstreamCount()).isEqualTo(0);
227+
assertThat(sink.currentSubscriberCount()).isEqualTo(0);
228228
}
229229

230230
private static io.lettuce.core.pubsub.api.reactive.ChannelMessage<ByteBuffer, ByteBuffer> createChannelMessage(

‎src/test/java/org/springframework/data/redis/connection/lettuce/LettuceSentinelIntegrationTests.java

+1
Original file line numberDiff line numberDiff line change
@@ -171,6 +171,7 @@ void shouldReadReplicasOfMastersCorrectly() {
171171
}
172172

173173
@Test // DATAREDIS-462
174+
@Disabled("Until Lettuce has moved to Sinks API")
174175
void factoryWorksWithoutClientResources() {
175176

176177
LettuceConnectionFactory factory = new LettuceConnectionFactory(SENTINEL_CONFIG);

‎src/test/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainerUnitTests.java

+27-24
Original file line numberDiff line numberDiff line change
@@ -20,10 +20,9 @@
2020
import static org.springframework.data.redis.util.ByteUtils.*;
2121

2222
import reactor.core.Disposable;
23-
import reactor.core.publisher.DirectProcessor;
2423
import reactor.core.publisher.Flux;
2524
import reactor.core.publisher.Mono;
26-
import reactor.core.publisher.MonoProcessor;
25+
import reactor.core.publisher.Sinks;
2726
import reactor.test.StepVerifier;
2827

2928
import java.nio.ByteBuffer;
@@ -92,8 +91,7 @@ void shouldSubscribeToMultiplePatterns() {
9291
container = createContainer();
9392

9493
container.receive(PatternTopic.of("foo*"), PatternTopic.of("bar*")).as(StepVerifier::create).thenRequest(1)
95-
.thenAwait()
96-
.thenCancel().verify();
94+
.thenAwait().thenCancel().verify();
9795

9896
verify(subscriptionMock).pSubscribe(getByteBuffer("foo*"), getByteBuffer("bar*"));
9997
}
@@ -124,15 +122,15 @@ void shouldSubscribeToMultipleChannels() {
124122
@Test // DATAREDIS-612
125123
void shouldEmitChannelMessage() {
126124

127-
DirectProcessor<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
125+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
128126

129-
when(subscriptionMock.receive()).thenReturn(processor);
127+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
130128
container = createContainer();
131129

132130
Flux<Message<String, String>> messageStream = container.receive(ChannelTopic.of("foo"));
133131

134132
messageStream.as(StepVerifier::create).then(() -> {
135-
processor.onNext(createChannelMessage("foo", "message"));
133+
sink.tryEmitNext(createChannelMessage("foo", "message"));
136134
}).assertNext(msg -> {
137135

138136
assertThat(msg.getChannel()).isEqualTo("foo");
@@ -143,15 +141,15 @@ void shouldEmitChannelMessage() {
143141
@Test // DATAREDIS-612
144142
void shouldEmitPatternMessage() {
145143

146-
DirectProcessor<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
144+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
147145

148-
when(subscriptionMock.receive()).thenReturn(processor);
146+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
149147
container = createContainer();
150148

151149
Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
152150

153151
messageStream.as(StepVerifier::create).then(() -> {
154-
processor.onNext(createPatternMessage("foo*", "foo", "message"));
152+
sink.tryEmitNext(createPatternMessage("foo*", "foo", "message"));
155153
}).assertNext(msg -> {
156154

157155
assertThat(msg.getPattern()).isEqualTo("foo*");
@@ -163,20 +161,22 @@ void shouldEmitPatternMessage() {
163161
@Test // DATAREDIS-612
164162
void shouldRegisterSubscription() {
165163

166-
MonoProcessor<Void> subscribeMono = MonoProcessor.create();
164+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().multicast().onBackpressureBuffer();
165+
166+
Sinks.One<Void> subscribeMono = Sinks.one();
167167

168168
reset(subscriptionMock);
169-
when(subscriptionMock.subscribe(any())).thenReturn(subscribeMono);
169+
when(subscriptionMock.subscribe(any())).thenReturn(subscribeMono.asMono());
170170
when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty());
171-
when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
171+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
172172
container = createContainer();
173173

174174
Flux<Message<String, String>> messageStream = container.receive(ChannelTopic.of("foo*"));
175175

176176
Disposable subscription = messageStream.subscribe();
177177

178178
assertThat(container.getActiveSubscriptions()).isEmpty();
179-
subscribeMono.onComplete();
179+
subscribeMono.tryEmitEmpty();
180180
assertThat(container.getActiveSubscriptions()).isNotEmpty();
181181
subscription.dispose();
182182
assertThat(container.getActiveSubscriptions()).isEmpty();
@@ -185,10 +185,12 @@ void shouldRegisterSubscription() {
185185
@Test // DATAREDIS-612, GH-1622
186186
void shouldRegisterSubscriptionMultipleSubscribers() {
187187

188+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().multicast().onBackpressureBuffer();
189+
188190
reset(subscriptionMock);
189191
when(subscriptionMock.subscribe(any())).thenReturn(Mono.empty());
190192
when(subscriptionMock.unsubscribe()).thenReturn(Mono.empty());
191-
when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
193+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
192194
container = createContainer();
193195

194196
Flux<Message<String, String>> messageStream = container.receive(new ChannelTopic("foo*"));
@@ -210,7 +212,8 @@ void shouldRegisterSubscriptionMultipleSubscribers() {
210212
@Test // DATAREDIS-612, GH-1622
211213
void shouldUnsubscribeOnCancel() {
212214

213-
when(subscriptionMock.receive()).thenReturn(DirectProcessor.create());
215+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
216+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
214217
container = createContainer();
215218

216219
Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
@@ -227,12 +230,12 @@ void shouldUnsubscribeOnCancel() {
227230
@Test // DATAREDIS-612
228231
void shouldTerminateSubscriptionsOnShutdown() {
229232

230-
DirectProcessor<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
233+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
231234

232-
when(subscriptionMock.receive()).thenReturn(processor);
235+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
233236
when(subscriptionMock.cancel()).thenReturn(Mono.defer(() -> {
234237

235-
processor.onError(new CancellationException());
238+
sink.tryEmitError(new CancellationException());
236239
return Mono.empty();
237240
}));
238241
container = createContainer();
@@ -247,19 +250,19 @@ void shouldTerminateSubscriptionsOnShutdown() {
247250
@Test // DATAREDIS-612
248251
void shouldCleanupDownstream() {
249252

250-
DirectProcessor<Message<ByteBuffer, ByteBuffer>> processor = DirectProcessor.create();
253+
Sinks.Many<Message<ByteBuffer, ByteBuffer>> sink = Sinks.many().unicast().onBackpressureBuffer();
251254

252-
when(subscriptionMock.receive()).thenReturn(processor);
255+
when(subscriptionMock.receive()).thenReturn(sink.asFlux());
253256
container = createContainer();
254257

255258
Flux<PatternMessage<String, String, String>> messageStream = container.receive(PatternTopic.of("foo*"));
256259

257260
messageStream.as(StepVerifier::create).then(() -> {
258-
assertThat(processor.hasDownstreams()).isTrue();
259-
processor.onNext(createPatternMessage("foo*", "foo", "message"));
261+
assertThat(sink.currentSubscriberCount()).isGreaterThan(0);
262+
sink.tryEmitNext(createPatternMessage("foo*", "foo", "message"));
260263
}).expectNextCount(1).thenCancel().verify();
261264

262-
assertThat(processor.hasDownstreams()).isFalse();
265+
assertThat(sink.currentSubscriberCount()).isEqualTo(0);
263266
}
264267

265268
private ReactiveRedisMessageListenerContainer createContainer() {

‎src/test/java/org/springframework/data/redis/test/extension/LettuceTestClientResources.java

+14-1
Original file line numberDiff line numberDiff line change
@@ -15,8 +15,11 @@
1515
*/
1616
package org.springframework.data.redis.test.extension;
1717

18+
import io.lettuce.core.event.Event;
19+
import io.lettuce.core.event.EventBus;
1820
import io.lettuce.core.resource.ClientResources;
1921
import io.lettuce.core.resource.DefaultClientResources;
22+
import reactor.core.publisher.Flux;
2023

2124
import java.util.concurrent.TimeUnit;
2225

@@ -33,7 +36,17 @@ public class LettuceTestClientResources {
3336

3437
static {
3538

36-
SHARED_CLIENT_RESOURCES = DefaultClientResources.builder()
39+
SHARED_CLIENT_RESOURCES = DefaultClientResources.builder().eventBus(new EventBus() {
40+
@Override
41+
public Flux<Event> get() {
42+
return Flux.empty();
43+
}
44+
45+
@Override
46+
public void publish(Event event) {
47+
48+
}
49+
})
3750
.build();
3851
ShutdownQueue.INSTANCE.register(() -> SHARED_CLIENT_RESOURCES.shutdown(0, 0, TimeUnit.MILLISECONDS));
3952
}

0 commit comments

Comments
 (0)