Improvement - Reactive method call cancellation support added

6 years ago · ad091fe14f
parent 7b20c0d9a5
commit ad091fe14f
1 changed files with 8 additions and 2 deletions
--- a/redisson/src/main/java/org/redisson/reactive/CommandReactiveService.java
+++ b/redisson/src/main/java/org/redisson/reactive/CommandReactiveService.java
@ -39,11 +39,17 @@ public class CommandReactiveService extends CommandAsyncService implements Comma
    public <R> Mono<R> reactive(Supplier<RFuture<R>> supplier) {
        return Flux.<R>create(emitter -> {
            emitter.onRequest(n -> {
-                supplier.get().whenComplete((v, e) -> {
+                RFuture<R> future = supplier.get();
+                future.whenComplete((v, e) -> {
                    if (e != null) {
                        emitter.error(e);
                        return;
-                    } 
+                    }
+
+                    emitter.onDispose(() -> {
+                        future.cancel(true);
+                    });
+                    
                    if (v != null) {
                        emitter.next(v);
                    }