Add Async Testkit Backend support

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 );
+    }
+}