Update TestkitRequestProcessorHandler to adhere to req/res pattern

This update aims to make sure the Teskit backend does not break the request / response pattern. For instance, by sending 2 responses together in case of internal callbacks.

Author: injectives

testkit-backend/src/main/java/neo4j/org/testkit/backend/channel/handler/TestkitRequestProcessorHandler.java

@@ -22,12 +22,15 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.ChannelInboundHandlerAdapter;
 import java.time.zone.ZoneRulesException;
+import java.util.ArrayDeque;
+import java.util.Queue;
 import java.util.concurrent.CompletableFuture;
 import java.util.concurrent.CompletionException;
 import java.util.concurrent.CompletionStage;
 import java.util.concurrent.Executor;
 import java.util.concurrent.Executors;
 import java.util.function.BiFunction;
+import java.util.function.Consumer;
 import neo4j.org.testkit.backend.CustomDriverError;
 import neo4j.org.testkit.backend.FrontendError;
 import neo4j.org.testkit.backend.TestkitState;
@@ -48,6 +51,7 @@ public class TestkitRequestProcessorHandler extends ChannelInboundHandlerAdapter
     // Some requests require multiple threads
     private final Executor requestExecutorService = Executors.newFixedThreadPool(10);
     private Channel channel;
+    private ResponseQueueHanlder responseQueueHanlder;
 
     public TestkitRequestProcessorHandler(BackendMode backendMode, Logging logging) {
         switch (backendMode) {
@@ -64,22 +68,22 @@ public TestkitRequestProcessorHandler(BackendMode backendMode, Logging logging)
     @Override
     public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
         channel = ctx.channel();
+        responseQueueHanlder = new ResponseQueueHanlder(channel::writeAndFlush);
         super.channelRegistered(ctx);
     }
 
     @Override
     public void channelRead(ChannelHandlerContext ctx, Object msg) {
         // Processing is done in a separate thread to avoid blocking EventLoop because some testing logic, like
         // resolvers support, is blocking.
+        responseQueueHanlder.setResponseReadyAndDispatchFirst();
         requestExecutorService.execute(() -> {
             try {
                 var request = (TestkitRequest) msg;
-                var responseStage = processorImpl.apply(request, testkitState);
-                responseStage.whenComplete((response, throwable) -> {
-                    if (throwable != null) {
-                        ctx.writeAndFlush(createErrorResponse(throwable));
-                    } else if (response != null) {
-                        ctx.writeAndFlush(response);
+                processorImpl.apply(request, testkitState).whenComplete((response, throwable) -> {
+                    var testkitResponse = throwable != null ? createErrorResponse(throwable) : response;
+                    if (testkitResponse != null) {
+                        responseQueueHanlder.offerAndDispatchFirst(testkitResponse);
                     }
                 });
             } catch (Throwable throwable) {
@@ -101,7 +105,8 @@ private static CompletionStage<TestkitResponse> wrapSyncRequest(
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        ctx.writeAndFlush(createErrorResponse(cause));
+        var response = createErrorResponse(cause);
+        responseQueueHanlder.offerAndDispatchFirst(response);
     }
 
     private TestkitResponse createErrorResponse(Throwable throwable) {
@@ -165,7 +170,7 @@ private void writeAndFlush(TestkitResponse response) {
         if (channel == null) {
             throw new IllegalStateException("Called before channel is initialized");
         }
-        channel.writeAndFlush(response);
+        responseQueueHanlder.offerAndDispatchFirst(response);
     }
 
     public enum BackendMode {
@@ -174,4 +179,34 @@ public enum BackendMode {
         REACTIVE_LEGACY,
         REACTIVE
     }
+
+    private static class ResponseQueueHanlder {
+        private final Consumer<TestkitResponse> responseWriter;
+        private final Queue<TestkitResponse> responseQueue = new ArrayDeque<>();
+        private boolean responseReady;
+
+        ResponseQueueHanlder(Consumer<TestkitResponse> responseWriter) {
+            this.responseWriter = responseWriter;
+        }
+
+        synchronized void setResponseReadyAndDispatchFirst() {
+            responseReady = true;
+            dispatchFirst();
+        }
+
+        synchronized void offerAndDispatchFirst(TestkitResponse response) {
+            responseQueue.offer(response);
+            if (responseReady) {
+                dispatchFirst();
+            }
+        }
+
+        private synchronized void dispatchFirst() {
+            var response = responseQueue.poll();
+            if (response != null) {
+                responseReady = false;
+                responseWriter.accept(responseQueue.poll());
+            }
+        }
+    }
 }