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

DATAREDIS-1053 - Polishing.

Browse Source 
Use early return and fix issue reference in tests.

Original Pull Request: #488
This commit is contained in:
Christoph Strobl
2019-10-29 09:51:21 +01:00
parent 6851186cf2
commit 9d62ead0e4
2 changed files with 23 additions and 25 deletions
Download Patch File Download Diff File Expand all files Collapse all files

45
src/main/java/org/springframework/data/redis/listener/ReactiveRedisMessageListenerContainer.java
View File

@@ -101,31 +101,30 @@ public class ReactiveRedisMessageListenerContainer implements DisposableBean {
private Mono<Void> doDestroy() {
ReactiveRedisConnection connection = this.connection;
if (connection != null) {
Flux<Void> terminationSignals = null;
while (!subscriptions.isEmpty()) {
Map<ReactiveSubscription, Subscribers> local = new HashMap<>(subscriptions);
List<Mono<Void>> monos = local.keySet().stream() //
.peek(subscriptions::remove) //
.map(ReactiveSubscription::cancel) //
.collect(Collectors.toList());
if (terminationSignals == null) {
terminationSignals = Flux.concat(monos);
} else {
terminationSignals = terminationSignals.mergeWith(Flux.concat(monos));
}
}
this.connection = null;
return terminationSignals != null ? terminationSignals.then(connection.closeLater()) : connection.closeLater();
if (this.connection == null) {
return Mono.empty();
}
return Mono.empty();
ReactiveRedisConnection connection = this.connection;
Flux<Void> terminationSignals = null;
while (!subscriptions.isEmpty()) {
Map<ReactiveSubscription, Subscribers> local = new HashMap<>(subscriptions);
List<Mono<Void>> monos = local.keySet().stream() //
.peek(subscriptions::remove) //
.map(ReactiveSubscription::cancel) //
.collect(Collectors.toList());
if (terminationSignals == null) {
terminationSignals = Flux.concat(monos);
} else {
terminationSignals = terminationSignals.mergeWith(Flux.concat(monos));
}
}
this.connection = null;
return terminationSignals != null ? terminationSignals.then(connection.closeLater()) : connection.closeLater();
}
/**

3
src/test/java/org/springframework/data/redis/core/ReactiveRedisTemplateUnitTests.java
View File

@@ -25,7 +25,6 @@ import reactor.test.StepVerifier;
import java.util.concurrent.atomic.AtomicBoolean;
import org.junit.Test;
import org.springframework.data.redis.connection.ReactivePubSubCommands;
import org.springframework.data.redis.connection.ReactiveRedisConnection;
import org.springframework.data.redis.connection.ReactiveRedisConnectionFactory;
@@ -59,7 +58,7 @@ public class ReactiveRedisTemplateUnitTests {
verifyNoMoreInteractions(connectionMock);
}
@Test // DATAREDIS-999
@Test // DATAREDIS-1053
public void listenToShouldSubscribeToChannel() {
AtomicBoolean closed = new AtomicBoolean();