[WIP] testing

Author: injectives

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

@@ -18,16 +18,23 @@
  */
 package neo4j.org.testkit.backend.channel.handler;
 
+import static org.neo4j.driver.internal.util.LockUtil.executeWithLock;
+
 import io.netty.channel.Channel;
 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.concurrent.locks.Lock;
+import java.util.concurrent.locks.ReentrantLock;
 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;
@@ -47,6 +54,9 @@ public class TestkitRequestProcessorHandler extends ChannelInboundHandlerAdapter
     private final BiFunction<TestkitRequest, TestkitState, CompletionStage<TestkitResponse>> processorImpl;
     // Some requests require multiple threads
     private final Executor requestExecutorService = Executors.newFixedThreadPool(10);
+    private final Lock lock = new ReentrantLock();
+    private final Queue<TestkitResponse> responseQueue = new ArrayDeque<>();
+    private boolean canDispatchResponse = false;
     private Channel channel;
 
     public TestkitRequestProcessorHandler(BackendMode backendMode, Logging logging) {
@@ -71,15 +81,26 @@ public void channelRegistered(ChannelHandlerContext ctx) throws Exception {
     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.
+        executeWithLock(lock, () -> {
+            canDispatchResponse = true;
+            trySending(ctx::writeAndFlush);
+        });
         requestExecutorService.execute(() -> {
             try {
                 var request = (TestkitRequest) msg;
                 var responseStage = processorImpl.apply(request, testkitState);
                 responseStage.whenComplete((response, throwable) -> {
                     if (throwable != null) {
-                        ctx.writeAndFlush(createErrorResponse(throwable));
+                        var errorResponse = createErrorResponse(throwable);
+                        executeWithLock(lock, () -> {
+                            responseQueue.offer(errorResponse);
+                            trySending(ctx::writeAndFlush);
+                        });
                     } else if (response != null) {
-                        ctx.writeAndFlush(response);
+                        executeWithLock(lock, () -> {
+                            responseQueue.offer(response);
+                            trySending(ctx::writeAndFlush);
+                        });
                     }
                 });
             } catch (Throwable throwable) {
@@ -88,6 +109,13 @@ public void channelRead(ChannelHandlerContext ctx, Object msg) {
         });
     }
 
+    private void trySending(Consumer<TestkitResponse> responseWriter) {
+        if (canDispatchResponse && !responseQueue.isEmpty()) {
+            canDispatchResponse = false;
+            responseWriter.accept(responseQueue.poll());
+        }
+    }
+
     private static CompletionStage<TestkitResponse> wrapSyncRequest(
             TestkitRequest testkitRequest, TestkitState testkitState) {
         var result = new CompletableFuture<TestkitResponse>();
@@ -101,7 +129,11 @@ private static CompletionStage<TestkitResponse> wrapSyncRequest(
 
     @Override
     public void exceptionCaught(ChannelHandlerContext ctx, Throwable cause) {
-        ctx.writeAndFlush(createErrorResponse(cause));
+        var response = createErrorResponse(cause);
+        executeWithLock(lock, () -> {
+            responseQueue.offer(response);
+            trySending(ctx::writeAndFlush);
+        });
     }
 
     private TestkitResponse createErrorResponse(Throwable throwable) {
@@ -165,7 +197,10 @@ private void writeAndFlush(TestkitResponse response) {
         if (channel == null) {
             throw new IllegalStateException("Called before channel is initialized");
         }
-        channel.writeAndFlush(response);
+        executeWithLock(lock, () -> {
+            responseQueue.offer(response);
+            trySending(channel::writeAndFlush);
+        });
     }
 
     public enum BackendMode {