Allow easier Bolt message versioning

Previously, decoding of inbound Bolt messages and their routing to the
state machine was very protocol V1 specific and not extensible. This
commit makes it possible to register protocol version specific message
deserializers. It also makes code always use same inbound message
classes. Two different implementations existed before - one for
production code and one for tests.

Central component is `BoltRequestMessageReader` which holds a
configurable set of message deserializers. Correct deserializer
is looked up by a message signature.

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/BoltIOException.java → community/bolt/src/main/java/org/neo4j/bolt/messaging/BoltIOException.java

@@ -17,7 +17,7 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.bolt.v1.messaging;
+package org.neo4j.bolt.messaging;
 
 import java.io.IOException;
 

community/bolt/src/main/java/org/neo4j/bolt/messaging/BoltRequestMessageReader.java

@@ -0,0 +1,101 @@
+/*
+ * Copyright (c) 2002-2018 "Neo4j,"
+ * Neo4j Sweden AB [http://neo4j.com]
+ *
+ * This file is part of Neo4j.
+ *
+ * Neo4j is free software: you can redistribute it and/or modify
+ * it under the terms of the GNU General Public License as published by
+ * the Free Software Foundation, either version 3 of the License, or
+ * (at your option) any later version.
+ *
+ * This program is distributed in the hope that it will be useful,
+ * but WITHOUT ANY WARRANTY; without even the implied warranty of
+ * MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.  See the
+ * GNU General Public License for more details.
+ *
+ * You should have received a copy of the GNU General Public License
+ * along with this program.  If not, see <http://www.gnu.org/licenses/>.
+ */
+package org.neo4j.bolt.messaging;
+
+import java.io.IOException;
+import java.util.List;
+import java.util.Map;
+
+import org.neo4j.bolt.logging.BoltMessageLogger;
+import org.neo4j.bolt.runtime.BoltConnection;
+import org.neo4j.bolt.runtime.BoltResponseHandler;
+import org.neo4j.bolt.runtime.Neo4jError;
+import org.neo4j.bolt.v1.packstream.PackStream;
+import org.neo4j.kernel.api.exceptions.Status;
+
+import static java.util.function.Function.identity;
+import static java.util.stream.Collectors.toMap;
+
+/**
+ * Reader for Bolt request messages made available via a {@link Neo4jPack.Unpacker}.
+ */
+public abstract class BoltRequestMessageReader
+{
+    private final BoltConnection connection;
+    private final BoltResponseHandler externalErrorResponseHandler;
+    private final BoltMessageLogger messageLogger;
+    private final Map<Integer,RequestMessageDecoder> decoders;
+
+    protected BoltRequestMessageReader( BoltConnection connection, BoltResponseHandler externalErrorResponseHandler,
+            List<RequestMessageDecoder> decoders, BoltMessageLogger messageLogger )
+    {
+        this.connection = connection;
+        this.externalErrorResponseHandler = externalErrorResponseHandler;
+        this.messageLogger = messageLogger;
+        this.decoders = decoders.stream().collect( toMap( RequestMessageDecoder::signature, identity() ) );
+    }
+
+    public void read( Neo4jPack.Unpacker unpacker ) throws IOException
+    {
+        try
+        {
+            doRead( unpacker );
+        }
+        catch ( BoltIOException e )
+        {
+            if ( e.causesFailureMessage() )
+            {
+                Neo4jError error = Neo4jError.from( e );
+                messageLogger.clientEvent( "ERROR", error::message );
+                connection.enqueue( stateMachine -> stateMachine.handleExternalFailure( error, externalErrorResponseHandler ) );
+            }
+            else
+            {
+                throw e;
+            }
+        }
+    }
+
+    private void doRead( Neo4jPack.Unpacker unpacker ) throws IOException
+    {
+        try
+        {
+            unpacker.unpackStructHeader();
+            int signature = unpacker.unpackStructSignature();
+
+            RequestMessageDecoder decoder = decoders.get( signature );
+            if ( decoder == null )
+            {
+                throw new BoltIOException( Status.Request.InvalidFormat,
+                        String.format( "Message 0x%s is not a valid message signature.", Integer.toHexString( signature ) ) );
+            }
+
+            RequestMessage message = decoder.decode( unpacker );
+            BoltResponseHandler resonseHandler = decoder.resonseHandler();
+
+            connection.enqueue( stateMachine -> stateMachine.process( message, resonseHandler ) );
+        }
+        catch ( PackStream.PackStreamException e )
+        {
+            throw new BoltIOException( Status.Request.InvalidFormat,
+                    String.format( "Unable to read message type. Error was: %s.", e.getMessage() ), e );
+        }
+    }
+}

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/Neo4jPack.java → community/bolt/src/main/java/org/neo4j/bolt/messaging/Neo4jPack.java

@@ -17,7 +17,7 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.bolt.v1.messaging;
+package org.neo4j.bolt.messaging;
 
 import java.io.IOException;
 

community/bolt/src/test/java/org/neo4j/bolt/v1/messaging/message/RequestMessage.java → community/bolt/src/main/java/org/neo4j/bolt/messaging/RequestMessage.java

@@ -17,11 +17,9 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.bolt.v1.messaging.message;
-
-import org.neo4j.bolt.v1.messaging.BoltRequestMessageHandler;
+package org.neo4j.bolt.messaging;
 
 public interface RequestMessage
 {
-    void dispatch( BoltRequestMessageHandler consumer );
+    boolean safeToProcessInAnyState();
 }

community/bolt/src/test/java/org/neo4j/bolt/v1/messaging/MessageRecorder.java → community/bolt/src/main/java/org/neo4j/bolt/messaging/RequestMessageDecoder.java

@@ -17,18 +17,17 @@
  * You should have received a copy of the GNU General Public License
  * along with this program.  If not, see <http://www.gnu.org/licenses/>.
  */
-package org.neo4j.bolt.v1.messaging;
+package org.neo4j.bolt.messaging;
 
-import java.util.ArrayList;
-import java.util.List;
+import java.io.IOException;
 
-public abstract class MessageRecorder<T>
+import org.neo4j.bolt.runtime.BoltResponseHandler;
+
+public interface RequestMessageDecoder
 {
-    protected List<T> messages = new ArrayList<>();
+    int signature();
 
-    public List<T> asList()
-    {
-        return messages;
-    }
+    BoltResponseHandler resonseHandler();
 
+    RequestMessage decode( Neo4jPack.Unpacker unpacker ) throws IOException;
 }

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltStateMachine.java

@@ -19,12 +19,13 @@
  */
 package org.neo4j.bolt.runtime;
 
+import org.neo4j.bolt.messaging.RequestMessage;
 import org.neo4j.internal.kernel.api.exceptions.KernelException;
 import org.neo4j.kernel.api.bolt.ManagedBoltStateMachine;
 
 public interface BoltStateMachine extends ManagedBoltStateMachine, AutoCloseable
 {
-    void process( StateMachineMessage message, BoltResponseHandler handler ) throws BoltConnectionFatality;
+    void process( RequestMessage message, BoltResponseHandler handler ) throws BoltConnectionFatality;
 
     boolean shouldStickOnThread();
 

community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltStateMachineState.java

@@ -19,9 +19,11 @@
  */
 package org.neo4j.bolt.runtime;
 
+import org.neo4j.bolt.messaging.RequestMessage;
+
 public interface BoltStateMachineState
 {
-    BoltStateMachineState process( StateMachineMessage message, StateMachineContext context ) throws BoltConnectionFatality;
+    BoltStateMachineState process( RequestMessage message, StateMachineContext context ) throws BoltConnectionFatality;
 
     String name();
 }

community/bolt/src/main/java/org/neo4j/bolt/transport/DefaultBoltProtocolPipelineInstaller.java

@@ -22,15 +22,15 @@
 import io.netty.channel.ChannelPipeline;
 
 import org.neo4j.bolt.BoltChannel;
+import org.neo4j.bolt.messaging.BoltRequestMessageReader;
+import org.neo4j.bolt.messaging.Neo4jPack;
 import org.neo4j.bolt.runtime.BoltConnection;
 import org.neo4j.bolt.transport.pipeline.ChunkDecoder;
 import org.neo4j.bolt.transport.pipeline.HouseKeeper;
 import org.neo4j.bolt.transport.pipeline.MessageAccumulator;
 import org.neo4j.bolt.transport.pipeline.MessageDecoder;
-import org.neo4j.bolt.v1.messaging.BoltMessageRouter;
-import org.neo4j.bolt.v1.messaging.BoltRequestMessageHandler;
+import org.neo4j.bolt.v1.messaging.BoltRequestMessageReaderV1;
 import org.neo4j.bolt.v1.messaging.BoltResponseMessageWriter;
-import org.neo4j.bolt.v1.messaging.Neo4jPack;
 import org.neo4j.kernel.impl.logging.LogService;
 
 /**
@@ -43,30 +43,26 @@ public class DefaultBoltProtocolPipelineInstaller implements BoltProtocolPipelin
 {
     private final BoltChannel boltChannel;
     private final Neo4jPack neo4jPack;
-    private final BoltResponseMessageWriter responseWriter;
-    private final BoltRequestMessageHandler messageHandler;
     private final LogService logging;
 
     private final BoltConnection connection;
 
-    public DefaultBoltProtocolPipelineInstaller( BoltChannel boltChannel, BoltConnection connection, Neo4jPack neo4jPack, TransportThrottleGroup throttleGroup,
-            LogService logging )
+    public DefaultBoltProtocolPipelineInstaller( BoltChannel boltChannel, BoltConnection connection, Neo4jPack neo4jPack, LogService logging )
     {
         this.boltChannel = boltChannel;
         this.connection = connection;
         this.neo4jPack = neo4jPack;
-        this.responseWriter = new BoltResponseMessageWriter( neo4jPack, connection.output(), logging, boltChannel.log() );
-        this.messageHandler = new BoltMessageRouter( logging.getInternalLog( getClass() ), boltChannel.log(), connection, responseWriter );
         this.logging = logging;
     }
 
+    @Override
     public void install()
     {
         ChannelPipeline pipeline = boltChannel.rawChannel().pipeline();
 
         pipeline.addLast( new ChunkDecoder() );
         pipeline.addLast( new MessageAccumulator() );
-        pipeline.addLast( new MessageDecoder( neo4jPack, messageHandler, logging ) );
+        pipeline.addLast( new MessageDecoder( neo4jPack, newRequestMessageReader(), logging ) );
         pipeline.addLast( new HouseKeeper( connection, logging ) );
     }
 
@@ -75,4 +71,10 @@ public long version()
     {
         return neo4jPack.version();
     }
+
+    private BoltRequestMessageReader newRequestMessageReader()
+    {
+        BoltResponseMessageWriter responseMessageWriter = new BoltResponseMessageWriter( neo4jPack, connection.output(), logging, boltChannel.log() );
+        return new BoltRequestMessageReaderV1( connection, responseMessageWriter, boltChannel.log(), logging );
+    }
 }

community/bolt/src/main/java/org/neo4j/bolt/transport/DefaultBoltProtocolPipelineInstallerFactory.java

@@ -20,9 +20,9 @@
 package org.neo4j.bolt.transport;
 
 import org.neo4j.bolt.BoltChannel;
+import org.neo4j.bolt.messaging.Neo4jPack;
 import org.neo4j.bolt.runtime.BoltConnection;
 import org.neo4j.bolt.runtime.BoltConnectionFactory;
-import org.neo4j.bolt.v1.messaging.Neo4jPack;
 import org.neo4j.bolt.v1.messaging.Neo4jPackV1;
 import org.neo4j.bolt.v2.messaging.Neo4jPackV2;
 import org.neo4j.kernel.impl.logging.LogService;
@@ -60,7 +60,7 @@ else if ( protocolVersion == Neo4jPackV2.VERSION )
 
     private BoltProtocolPipelineInstaller newProtocolPipelineInstaller( BoltChannel channel, Neo4jPack neo4jPack )
     {
-        return new DefaultBoltProtocolPipelineInstaller( channel, newBoltConnection( channel ), neo4jPack, throttleGroup, logService );
+        return new DefaultBoltProtocolPipelineInstaller( channel, newBoltConnection( channel ), neo4jPack, logService );
     }
 
     private BoltConnection newBoltConnection( BoltChannel channel )

community/bolt/src/main/java/org/neo4j/bolt/transport/pipeline/MessageDecoder.java

@@ -23,11 +23,8 @@
 import io.netty.channel.ChannelHandlerContext;
 import io.netty.channel.SimpleChannelInboundHandler;
 
-import org.neo4j.bolt.runtime.Neo4jError;
-import org.neo4j.bolt.v1.messaging.BoltIOException;
-import org.neo4j.bolt.v1.messaging.BoltRequestMessageHandler;
-import org.neo4j.bolt.v1.messaging.BoltRequestMessageReader;
-import org.neo4j.bolt.v1.messaging.Neo4jPack;
+import org.neo4j.bolt.messaging.BoltRequestMessageReader;
+import org.neo4j.bolt.messaging.Neo4jPack;
 import org.neo4j.bolt.v1.packstream.ByteBufInput;
 import org.neo4j.kernel.impl.logging.LogService;
 import org.neo4j.logging.Log;
@@ -37,15 +34,15 @@
 public class MessageDecoder extends SimpleChannelInboundHandler<ByteBuf>
 {
     private final ByteBufInput input;
+    private final Neo4jPack.Unpacker unpacker;
     private final BoltRequestMessageReader reader;
-    private final BoltRequestMessageHandler messageHandler;
     private final Log log;
 
-    public MessageDecoder( Neo4jPack pack, BoltRequestMessageHandler messageHandler, LogService logService )
+    public MessageDecoder( Neo4jPack pack, BoltRequestMessageReader reader, LogService logService )
     {
         this.input = new ByteBufInput();
-        this.reader = new BoltRequestMessageReader( pack.newUnpacker( input ) );
-        this.messageHandler = messageHandler;
+        this.unpacker = pack.newUnpacker( input );
+        this.reader = reader;
         this.log = logService.getInternalLog( getClass() );
     }
 
@@ -56,19 +53,7 @@ protected void channelRead0( ChannelHandlerContext channelHandlerContext, ByteBu
         byteBuf.markReaderIndex();
         try
         {
-            reader.read( messageHandler );
-        }
-        catch ( BoltIOException ex )
-        {
-            if ( ex.causesFailureMessage() )
-            {
-                messageHandler.onExternalError( Neo4jError.from( ex ) );
-            }
-            else
-            {
-                logMessageOnError( byteBuf );
-                throw ex;
-            }
+            reader.read( unpacker );
         }
         catch ( Throwable error )
         {