Add Async Testkit Backend support

This update brings async Testkit backend support that uses async driver for testing and is compatible with the current Testkit protocol. This allows async driver testing using the same Testkit tests transparently.

Author: injectives

testkit-backend/pom.xml

@@ -25,6 +25,10 @@
             <artifactId>neo4j-java-driver</artifactId>
             <version>${project.version}</version>
         </dependency>
+        <dependency>
+            <groupId>io.netty</groupId>
+            <artifactId>netty-handler</artifactId>
+        </dependency>
         <dependency>
             <groupId>com.fasterxml.jackson.core</groupId>
             <artifactId>jackson-core</artifactId>

testkit-backend/src/main/java/neo4j/org/testkit/backend/AsyncBackendServer.java

@@ -0,0 +1,63 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package neo4j.org.testkit.backend;
+
+import io.netty.bootstrap.ServerBootstrap;
+import io.netty.channel.ChannelFuture;
+import io.netty.channel.ChannelInitializer;
+import io.netty.channel.EventLoopGroup;
+import io.netty.channel.nio.NioEventLoopGroup;
+import io.netty.channel.socket.SocketChannel;
+import io.netty.channel.socket.nio.NioServerSocketChannel;
+import neo4j.org.testkit.backend.channel.handler.TestkitMessageInboundHandler;
+import neo4j.org.testkit.backend.channel.handler.TestkitMessageOutboundHandler;
+import neo4j.org.testkit.backend.channel.handler.TestkitRequestProcessorHandler;
+import neo4j.org.testkit.backend.channel.handler.TestkitRequestResponseMapperHandler;
+
+public class AsyncBackendServer
+{
+    public void run() throws InterruptedException
+    {
+        EventLoopGroup group = new NioEventLoopGroup();
+        try
+        {
+            ServerBootstrap bootstrap = new ServerBootstrap();
+            bootstrap.group( group )
+                     .channel( NioServerSocketChannel.class )
+                     .localAddress( 9876 )
+                     .childHandler( new ChannelInitializer<SocketChannel>()
+                     {
+                         @Override
+                         protected void initChannel( SocketChannel channel )
+                         {
+                             channel.pipeline().addLast( new TestkitMessageInboundHandler() );
+                             channel.pipeline().addLast( new TestkitMessageOutboundHandler() );
+                             channel.pipeline().addLast( new TestkitRequestResponseMapperHandler() );
+                             channel.pipeline().addLast( new TestkitRequestProcessorHandler() );
+                         }
+                     } );
+            ChannelFuture server = bootstrap.bind().sync();
+            server.channel().closeFuture().sync();
+        }
+        finally
+        {
+            group.shutdownGracefully().sync();
+        }
+    }
+}

testkit-backend/src/main/java/neo4j/org/testkit/backend/AsyncSessionState.java

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package neo4j.org.testkit.backend;
+
+import lombok.Getter;
+import lombok.Setter;
+
+import java.util.concurrent.CompletableFuture;
+
+import org.neo4j.driver.async.AsyncSession;
+
+@Getter
+@Setter
+public class AsyncSessionState
+{
+    public AsyncSession session;
+    public CompletableFuture<Void> txWorkFuture;
+
+    public AsyncSessionState( AsyncSession session )
+    {
+        this.session = session;
+    }
+}

testkit-backend/src/main/java/neo4j/org/testkit/backend/BackendServer.java

@@ -0,0 +1,75 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package neo4j.org.testkit.backend;
+
+import java.io.BufferedReader;
+import java.io.BufferedWriter;
+import java.io.IOException;
+import java.io.InputStreamReader;
+import java.io.OutputStreamWriter;
+import java.io.UncheckedIOException;
+import java.net.ServerSocket;
+import java.net.Socket;
+import java.util.concurrent.CompletableFuture;
+
+public class BackendServer
+{
+    public void run() throws IOException
+    {
+        ServerSocket serverSocket = new ServerSocket( 9876 );
+
+        System.out.println( "Java TestKit Backend Started on port: " + serverSocket.getLocalPort() );
+
+        while ( true )
+        {
+            final Socket clientSocket = serverSocket.accept();
+            CompletableFuture.runAsync( () -> handleClient( clientSocket ) );
+        }
+    }
+
+    private void handleClient( Socket clientSocket )
+    {
+        try
+        {
+            System.out.println( "Handling connection from: " + clientSocket.getRemoteSocketAddress() );
+            BufferedReader in = new BufferedReader( new InputStreamReader( clientSocket.getInputStream() ) );
+            BufferedWriter out = new BufferedWriter( new OutputStreamWriter( clientSocket.getOutputStream() ) );
+            CommandProcessor commandProcessor = new CommandProcessor( in, out );
+
+            boolean cont = true;
+            while ( cont )
+            {
+                try
+                {
+                    cont = commandProcessor.process();
+                }
+                catch ( Exception e )
+                {
+                    e.printStackTrace();
+                    clientSocket.close();
+                    cont = false;
+                }
+            }
+        }
+        catch ( IOException ex )
+        {
+            throw new UncheckedIOException( ex );
+        }
+    }
+}

testkit-backend/src/main/java/neo4j/org/testkit/backend/Runner.java

@@ -18,58 +18,19 @@
  */
 package neo4j.org.testkit.backend;
 
-import java.io.BufferedReader;
-import java.io.BufferedWriter;
 import java.io.IOException;
-import java.io.InputStreamReader;
-import java.io.OutputStreamWriter;
-import java.io.UncheckedIOException;
-import java.net.ServerSocket;
-import java.net.Socket;
-import java.util.concurrent.CompletableFuture;
 
 public class Runner
 {
-    public static void main( String[] args ) throws IOException
+    public static void main( String[] args ) throws IOException, InterruptedException
     {
-        ServerSocket serverSocket = new ServerSocket( 9876 );
-
-        System.out.println( "Java TestKit Backend Started on port: " + serverSocket.getLocalPort() );
-
-        while ( true )
-        {
-            final Socket clientSocket = serverSocket.accept();
-            CompletableFuture.runAsync( () -> handleClient( clientSocket ) );
-        }
-    }
-
-    private static void handleClient( Socket clientSocket )
-    {
-        try
+        if ( args.length > 0 && args[0].equals( "async" ) )
         {
-            System.out.println( "Handling connection from: " + clientSocket.getRemoteSocketAddress() );
-            BufferedReader in = new BufferedReader( new InputStreamReader( clientSocket.getInputStream() ) );
-            BufferedWriter out = new BufferedWriter( new OutputStreamWriter( clientSocket.getOutputStream() ) );
-            CommandProcessor commandProcessor = new CommandProcessor( in, out );
-
-            boolean cont = true;
-            while ( cont )
-            {
-                try
-                {
-                    cont = commandProcessor.process();
-                }
-                catch ( Exception e )
-                {
-                    e.printStackTrace();
-                    clientSocket.close();
-                    cont = false;
-                }
-            }
+            new AsyncBackendServer().run();
         }
-        catch ( IOException ex )
+        else
         {
-            throw new UncheckedIOException( ex );
+            new BackendServer().run();
         }
     }
 }

testkit-backend/src/main/java/neo4j/org/testkit/backend/TestkitState.java

@@ -31,6 +31,8 @@
 import org.neo4j.driver.Driver;
 import org.neo4j.driver.Result;
 import org.neo4j.driver.Transaction;
+import org.neo4j.driver.async.AsyncTransaction;
+import org.neo4j.driver.async.ResultCursor;
 import org.neo4j.driver.exceptions.Neo4jException;
 import org.neo4j.driver.internal.cluster.RoutingTableRegistry;
 import org.neo4j.driver.net.ServerAddress;
@@ -41,8 +43,11 @@ public class TestkitState
     private final Map<String,Driver> drivers = new HashMap<>();
     private final Map<String,RoutingTableRegistry> routingTableRegistry = new HashMap<>();
     private final Map<String,SessionState> sessionStates = new HashMap<>();
+    private final Map<String,AsyncSessionState> asyncSessionStates = new HashMap<>();
     private final Map<String,Result> results = new HashMap<>();
+    private final Map<String,ResultCursor> resultCursors = new HashMap<>();
     private final Map<String,Transaction> transactions = new HashMap<>();
+    private final Map<String,AsyncTransaction> asyncTransactions = new HashMap<>();
     private final Map<String,Neo4jException> errors = new HashMap<>();
     private int idGenerator = 0;
     private final Consumer<TestkitResponse> responseWriter;

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

@@ -0,0 +1,81 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package neo4j.org.testkit.backend.channel.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.SimpleChannelInboundHandler;
+import io.netty.util.CharsetUtil;
+
+import java.util.ArrayList;
+import java.util.List;
+import java.util.Optional;
+
+public class TestkitMessageInboundHandler extends SimpleChannelInboundHandler<ByteBuf>
+{
+    private final StringBuilder requestBuffer = new StringBuilder();
+
+    @Override
+    public void channelRead0( ChannelHandlerContext ctx, ByteBuf byteBuf )
+    {
+        String requestStr = byteBuf.toString( CharsetUtil.UTF_8 );
+        requestBuffer.append( requestStr );
+
+        List<String> testkitMessages = new ArrayList<>();
+        Optional<String> testkitMessageOpt = extractTestkitMessage();
+        while ( testkitMessageOpt.isPresent() )
+        {
+            testkitMessages.add( testkitMessageOpt.get() );
+            testkitMessageOpt = extractTestkitMessage();
+        }
+
+        testkitMessages.forEach( ctx::fireChannelRead );
+    }
+
+    private Optional<String> extractTestkitMessage()
+    {
+        String requestEndMarker = "#request end\n";
+        int endMarkerIndex = requestBuffer.indexOf( requestEndMarker );
+        if ( endMarkerIndex < 0 )
+        {
+            return Optional.empty();
+        }
+        String requestBeginMarker = "#request begin\n";
+        int beginMarkerIndex = requestBuffer.indexOf( requestBeginMarker );
+        if ( beginMarkerIndex != 0 )
+        {
+            throw new RuntimeException( "Unexpected data in message buffer" );
+        }
+        // extract Testkit message without markers
+        String testkitMessage = requestBuffer.substring( requestBeginMarker.length(), endMarkerIndex );
+        if ( testkitMessage.contains( requestBeginMarker ) || testkitMessage.contains( requestEndMarker ) )
+        {
+            throw new RuntimeException( "Testkit message contains request markers" );
+        }
+        // remove Testkit message from buffer
+        requestBuffer.delete( 0, endMarkerIndex + requestEndMarker.length() + 1 );
+        return Optional.of( testkitMessage );
+    }
+
+    @Override
+    public void exceptionCaught( ChannelHandlerContext ctx, Throwable cause )
+    {
+        ctx.close();
+    }
+}

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

@@ -0,0 +1,39 @@
+/*
+ * Copyright (c) "Neo4j"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *     http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package neo4j.org.testkit.backend.channel.handler;
+
+import io.netty.buffer.ByteBuf;
+import io.netty.buffer.Unpooled;
+import io.netty.channel.ChannelHandlerContext;
+import io.netty.channel.ChannelOutboundHandlerAdapter;
+import io.netty.channel.ChannelPromise;
+
+import java.nio.charset.StandardCharsets;
+
+public class TestkitMessageOutboundHandler extends ChannelOutboundHandlerAdapter
+{
+    @Override
+    public void write( ChannelHandlerContext ctx, Object msg, ChannelPromise promise )
+    {
+        String testkitResponseStr = (String) msg;
+        String testkitMessage = String.format( "#response begin\n%s\n#response end\n", testkitResponseStr );
+        ByteBuf byteBuf = Unpooled.copiedBuffer( testkitMessage, StandardCharsets.UTF_8 );
+        ctx.writeAndFlush( byteBuf, promise );
+    }
+}