Remove UnicastProcessor from reply router when timeout triggers

Author: dbuos

async/async-commons/src/main/java/org/reactivecommons/async/impl/RabbitDirectAsyncGateway.java

@@ -17,6 +17,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.UUID;
+import java.util.concurrent.TimeoutException;
 
 import static java.lang.Boolean.TRUE;
 import static org.reactivecommons.async.impl.Headers.*;
@@ -62,6 +63,7 @@ public <T, R> Mono<R> requestReply(AsyncQuery<T> query, String targetName, Class
 
         final Mono<R> replyHolder = router.register(correlationID)
                 .timeout(replyTimeout)
+                .doOnError(TimeoutException.class, _t -> router.deregister(correlationID))
                 .flatMap(s -> fromCallable(() -> converter.readValue(s, type)));
 
         Map<String, Object> headers = new HashMap<>();

async/async-commons/src/main/java/org/reactivecommons/async/impl/reply/ReactiveReplyRouter.java

@@ -24,6 +24,10 @@ public void routeReply(String correlationID, Message data) {
         }
     }
 
+    public void deregister(String correlationID){
+        processors.remove(correlationID);
+    }
+
     public <E> void routeError(String correlationID, String data) {
         final UnicastProcessor<Message> processor = processors.remove(correlationID);
         if (processor != null) {

async/async-commons/src/test/java/org/reactivecommons/async/impl/RabbitDirectAsyncGatewayTest.java

@@ -28,11 +28,13 @@
 import reactor.test.StepVerifier;
 import reactor.util.concurrent.Queues;
 
+import java.time.Duration;
 import java.util.List;
 import java.util.Map;
 import java.util.UUID;
 import java.util.concurrent.Semaphore;
 import java.util.concurrent.ThreadLocalRandom;
+import java.util.concurrent.TimeoutException;
 import java.util.stream.Collectors;
 import java.util.stream.IntStream;
 
@@ -58,6 +60,25 @@ public void init(ReactiveMessageSender sender) {
         asyncGateway = new RabbitDirectAsyncGateway(config, router, sender, "exchange", converter);
     }
 
+    @Test
+    public void shouldReleaseRouterResourcesOnTimeout(){
+        BrokerConfig config = new BrokerConfig(false, false, false, Duration.ofSeconds(1));
+        asyncGateway = new RabbitDirectAsyncGateway(config, router, senderMock, "ex", converter);
+        when(router.register(anyString())).thenReturn(Mono.never());
+        when(senderMock.sendNoConfirm(any(), anyString(), anyString(), anyMap(), anyBoolean()))
+            .thenReturn(Mono.empty());
+
+        AsyncQuery<String> query = new AsyncQuery<>("some.query", "data");
+        asyncGateway.requestReply(query, "some.target", String.class)
+            .as(StepVerifier::create)
+            .expectError(TimeoutException.class)
+            .verify();
+
+        ArgumentCaptor<String> captor = ArgumentCaptor.forClass(String.class);
+        verify(router).register(captor.capture());
+        verify(router).deregister(captor.getValue());
+    }
+
     @Test
     public void shouldSendInOptimalTime() throws InterruptedException {
         init(getReactiveMessageSender());

gradle.properties

@@ -1,4 +1,4 @@
-version=1.0.0-beta2
+version=1.0.0-beta3
 springBootVersion=2.2.9.RELEASE
 gradleVersionsVersion=0.28.0
 reactorRabbitVersion=1.5.0