Fix ReactiveRedisMessageListenerContainer disposal via ReactiveRedisTemplate.listenToLater(…).

mp911de · christophstrobl · commit 7b6a697265bd · 2023-01-13T07:43:37.000+01:00
We now properly dispose the container at the right time and not as result of the Mono<…> completion. Closes: #2386 Original Pull Request: #2467
diff --git a/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java b/src/main/java/org/springframework/data/redis/core/ReactiveRedisTemplate.java
@@ -291,7 +291,8 @@ public Mono<Flux<? extends Message<String, V>>> listenToLater(Topic... topics) {
 
 		return (Mono) container.receiveLater(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(),
 						getSerializationContext().getValueSerializationPair()) //
-				.doFinally((signalType) -> container.destroyLater().subscribe());
+				.map(it -> it.doFinally(signalType -> container.destroyLater().subscribe()))
+				.doOnCancel(() -> container.destroyLater().subscribe());
 	}
 
 	// -------------------------------------------------------------------------

Original file line number	Diff line number	Diff line change
`@@ -291,7 +291,8 @@ public Mono<Flux<? extends Message<String, V>>> listenToLater(Topic... topics) {`
`291`	`291`
`292`	`292`	`return (Mono) container.receiveLater(Arrays.asList(topics), getSerializationContext().getStringSerializationPair(),`
`293`	`293`	`getSerializationContext().getValueSerializationPair()) //`
`294`		`- .doFinally((signalType) -> container.destroyLater().subscribe());`
	`294`	`+ .map(it -> it.doFinally(signalType -> container.destroyLater().subscribe()))`
	`295`	`+ .doOnCancel(() -> container.destroyLater().subscribe());`
`295`	`296`	`}`
`296`	`297`
`297`	`298`	`// -------------------------------------------------------------------------`