Added Hello message

to replace old init message.
The new Hello message follow the following format:

**Message format**:  HELLO {metadata}

**Request Metadata Keys**
user_agent
keys used for authToken

**Response Metadata Keys on SUCCESS**
server: "Neo4j/3.4.0"
routing_table: "dbms.cluster.routing.getRoutingTable"
connection: <UUID>

Close conn on FAILURE

community/bolt/src/main/java/org/neo4j/bolt/BoltServer.java

@@ -41,8 +41,8 @@
 import org.neo4j.bolt.transport.NettyServer.ProtocolInitializer;
 import org.neo4j.bolt.transport.SocketTransport;
 import org.neo4j.bolt.transport.TransportThrottleGroup;
-import org.neo4j.bolt.v1.runtime.BoltStateMachineFactory;
-import org.neo4j.bolt.v1.runtime.BoltStateMachineFactoryImpl;
+import org.neo4j.bolt.runtime.BoltStateMachineFactory;
+import org.neo4j.bolt.runtime.BoltStateMachineFactoryImpl;
 import org.neo4j.graphdb.DependencyResolver;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
 import org.neo4j.helpers.ListenSocketAddress;

community/bolt/src/main/java/org/neo4j/bolt/logging/BoltMessageLogger.java

@@ -40,7 +40,7 @@ public interface BoltMessageLogger
 
     void serverError( String eventName, Status status );
 
-    void logInit( String userAgent );
+    void logUserAgent( String userAgent );
 
     void logRun();
 

community/bolt/src/main/java/org/neo4j/bolt/logging/BoltMessageLoggerImpl.java

@@ -102,7 +102,7 @@ public void serverError( String eventName, Status status )
     }
 
     @Override
-    public void logInit( String userAgent )
+    public void logUserAgent( String userAgent )
     {
         clientEvent( "INIT", () -> userAgent);
     }

community/bolt/src/main/java/org/neo4j/bolt/logging/NullBoltMessageLogger.java

@@ -73,7 +73,7 @@ public void serverError( String eventName, Status status )
     }
 
     @Override
-    public void logInit( String userAgent )
+    public void logUserAgent( String userAgent )
     {
     }
 

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

@@ -47,4 +47,6 @@ public interface BoltStateMachine extends ManagedBoltStateMachine, AutoCloseable
 
     @Override
     void close();
+
+    String id();
 }

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/BoltStateMachineFactory.java → community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltStateMachineFactory.java

@@ -17,10 +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.runtime;
+package org.neo4j.bolt.runtime;
 
 import org.neo4j.bolt.BoltChannel;
-import org.neo4j.bolt.runtime.BoltStateMachine;
 
 /**
  * Factory class for Bolt runtime environments.

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/BoltStateMachineFactoryImpl.java → community/bolt/src/main/java/org/neo4j/bolt/runtime/BoltStateMachineFactoryImpl.java

@@ -17,16 +17,19 @@
  * 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.runtime;
+package org.neo4j.bolt.runtime;
 
 import java.time.Clock;
 import java.time.Duration;
 
 import org.neo4j.bolt.BoltChannel;
-import org.neo4j.bolt.runtime.BoltStateMachine;
-import org.neo4j.bolt.runtime.BoltStateMachineSPI;
-import org.neo4j.bolt.runtime.TransactionStateMachineSPI;
 import org.neo4j.bolt.security.auth.Authentication;
+import org.neo4j.bolt.v1.BoltProtocolV1;
+import org.neo4j.bolt.v1.runtime.BoltStateMachineV1;
+import org.neo4j.bolt.v1.runtime.BoltStateMachineV1SPI;
+import org.neo4j.bolt.v2.BoltProtocolV2;
+import org.neo4j.bolt.v3.BoltProtocolV3;
+import org.neo4j.bolt.v3.BoltStateMachineV3;
 import org.neo4j.graphdb.factory.GraphDatabaseSettings;
 import org.neo4j.kernel.AvailabilityGuard;
 import org.neo4j.kernel.api.bolt.BoltConnectionTracker;
@@ -62,7 +65,18 @@ public BoltStateMachineFactoryImpl( GraphDatabaseAPI db, UsageData usageData, Av
     @Override
     public BoltStateMachine newStateMachine( long protocolVersion, BoltChannel boltChannel )
     {
-        return newStateMachineV1( boltChannel );
+        if ( protocolVersion == BoltProtocolV1.VERSION || protocolVersion == BoltProtocolV2.VERSION )
+        {
+            return newStateMachineV1( boltChannel );
+        }
+        else if ( protocolVersion == BoltProtocolV3.VERSION )
+        {
+            return newStateMachineV3( boltChannel );
+        }
+        else
+        {
+            return null;
+        }
     }
 
     private BoltStateMachine newStateMachineV1( BoltChannel boltChannel )
@@ -72,6 +86,13 @@ private BoltStateMachine newStateMachineV1( BoltChannel boltChannel )
         return new BoltStateMachineV1( boltSPI, boltChannel, clock );
     }
 
+    private BoltStateMachine newStateMachineV3( BoltChannel boltChannel )
+    {
+        TransactionStateMachineSPI transactionSPI = createTxSpi( clock );
+        BoltStateMachineSPI boltSPI = new BoltStateMachineV1SPI( boltChannel, usageData, logging, authentication, connectionTracker, transactionSPI );
+        return new BoltStateMachineV3( boltSPI, boltChannel, clock );
+    }
+
     private TransactionStateMachineSPI createTxSpi( Clock clock )
     {
         long bookmarkReadyTimeout = config.get( GraphDatabaseSettings.bookmark_ready_timeout ).toMillis();

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

@@ -27,6 +27,8 @@
 import org.neo4j.kernel.impl.logging.LogService;
 import org.neo4j.kernel.monitoring.Monitors;
 
+import static java.util.Objects.requireNonNull;
+
 public class DefaultBoltConnectionFactory implements BoltConnectionFactory
 {
     private final BoltSchedulerProvider schedulerProvider;
@@ -52,6 +54,9 @@ public DefaultBoltConnectionFactory( BoltSchedulerProvider schedulerProvider, Tr
     @Override
     public BoltConnection newConnection( BoltChannel channel, BoltStateMachine stateMachine )
     {
+        requireNonNull( channel );
+        requireNonNull( stateMachine );
+
         BoltScheduler scheduler = schedulerProvider.get( channel );
         BoltConnectionQueueMonitor connectionQueueMonitor =
                 queueMonitor == null ? scheduler : new BoltConnectionQueueMonitorAggregate( scheduler, queueMonitor );

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

@@ -34,4 +34,6 @@ public interface StateMachineContext
     MutableConnectionState connectionState();
 
     Clock clock();
+
+    String connectionId();
 }

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

@@ -19,12 +19,15 @@
  */
 package org.neo4j.bolt.transport;
 
+import java.io.IOException;
+
 import org.neo4j.bolt.BoltChannel;
 import org.neo4j.bolt.BoltProtocol;
 import org.neo4j.bolt.runtime.BoltConnectionFactory;
+import org.neo4j.bolt.runtime.BoltStateMachineFactory;
 import org.neo4j.bolt.v1.BoltProtocolV1;
-import org.neo4j.bolt.v1.runtime.BoltStateMachineFactory;
 import org.neo4j.bolt.v2.BoltProtocolV2;
+import org.neo4j.bolt.v3.BoltProtocolV3;
 import org.neo4j.kernel.impl.logging.LogService;
 
 public class DefaultBoltProtocolFactory implements BoltProtocolFactory
@@ -52,6 +55,10 @@ else if ( protocolVersion == BoltProtocolV2.VERSION )
         {
             return new BoltProtocolV2( channel, connectionFactory, stateMachineFactory, logService );
         }
+        else if ( protocolVersion == BoltProtocolV3.VERSION )
+        {
+            return new BoltProtocolV3( channel, connectionFactory, stateMachineFactory, logService );
+        }
         else
         {
             return null;

community/bolt/src/main/java/org/neo4j/bolt/v1/BoltProtocolV1.java

@@ -28,14 +28,14 @@
 import org.neo4j.bolt.runtime.BoltConnection;
 import org.neo4j.bolt.runtime.BoltConnectionFactory;
 import org.neo4j.bolt.runtime.BoltStateMachine;
+import org.neo4j.bolt.runtime.BoltStateMachineFactory;
 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.BoltRequestMessageReaderV1;
 import org.neo4j.bolt.v1.messaging.BoltResponseMessageWriterV1;
 import org.neo4j.bolt.v1.messaging.Neo4jPackV1;
-import org.neo4j.bolt.v1.runtime.BoltStateMachineFactory;
 import org.neo4j.kernel.impl.logging.LogService;
 
 /**
@@ -61,7 +61,7 @@ public BoltProtocolV1( BoltChannel channel, BoltConnectionFactory connectionFact
         this.connection = connectionFactory.newConnection( channel, stateMachine );
 
         this.neo4jPack = createPack();
-        this.messageReader = createBoltMessageReaderV1( channel, neo4jPack, connection, logging );
+        this.messageReader = createMessageReader( channel, neo4jPack, connection, logging );
     }
 
     /**
@@ -89,7 +89,7 @@ public long version()
         return VERSION;
     }
 
-    public static BoltRequestMessageReader createBoltMessageReaderV1( BoltChannel channel, Neo4jPack neo4jPack, BoltConnection connection, LogService logging )
+    protected BoltRequestMessageReader createMessageReader( BoltChannel channel, Neo4jPack neo4jPack, BoltConnection connection, LogService logging )
     {
         BoltResponseMessageWriterV1 responseWriter = new BoltResponseMessageWriterV1( neo4jPack, connection.output(), logging, channel.log() );
         return new BoltRequestMessageReaderV1( connection, responseWriter, channel.log(), logging );

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/BoltStateMachineV1Context.java

@@ -77,4 +77,10 @@ public Clock clock()
     {
         return clock;
     }
+
+    @Override
+    public String connectionId()
+    {
+        return machine.id();
+    }
 }

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/MessageProcessingHandler.java

@@ -40,7 +40,7 @@
 
 import static org.neo4j.bolt.v1.messaging.response.IgnoredMessage.IGNORED_MESSAGE;
 
-class MessageProcessingHandler implements BoltResponseHandler
+public class MessageProcessingHandler implements BoltResponseHandler
 {
     // Errors that are expected when the client disconnects mid-operation
     private static final Set<Status> CLIENT_MID_OP_DISCONNECT_ERRORS =
@@ -54,7 +54,7 @@ class MessageProcessingHandler implements BoltResponseHandler
     private Neo4jError error;
     private boolean ignored;
 
-    MessageProcessingHandler( BoltResponseMessageWriter messageWriter, BoltConnection connection, Log logger )
+    public MessageProcessingHandler( BoltResponseMessageWriter messageWriter, BoltConnection connection, Log logger )
     {
         this.messageWriter = messageWriter;
         this.connection = connection;

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/ResultHandler.java

@@ -27,9 +27,9 @@
 import org.neo4j.logging.Log;
 import org.neo4j.values.AnyValue;
 
-class ResultHandler extends MessageProcessingHandler
+public class ResultHandler extends MessageProcessingHandler
 {
-    ResultHandler( BoltResponseMessageWriter handler, BoltConnection connection, Log log )
+    public ResultHandler( BoltResponseMessageWriter handler, BoltConnection connection, Log log )
     {
         super( handler, connection, log );
     }

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/decoder/InitMessageDecoder.java

@@ -59,7 +59,7 @@ public RequestMessage decode( Neo4jPack.Unpacker unpacker ) throws IOException
     {
         String userAgent = unpacker.unpackString();
         Map<String,Object> authToken = readAuthToken( unpacker );
-        messageLogger.logInit( userAgent );
+        messageLogger.logUserAgent( userAgent );
         return new InitMessage( userAgent, authToken );
     }
 

community/bolt/src/main/java/org/neo4j/bolt/v1/messaging/decoder/PrimitiveOnlyValueWriter.java

@@ -37,9 +37,9 @@
  * {@link AnyValueWriter Writer} that allows to convert {@link AnyValue} to any primitive Java type. It explicitly
  * prohibits conversion of nodes, relationships, spatial and temporal types. They are not expected in auth token map.
  */
-class PrimitiveOnlyValueWriter extends BaseToObjectValueWriter<RuntimeException>
+public class PrimitiveOnlyValueWriter extends BaseToObjectValueWriter<RuntimeException>
 {
-    Object valueAsObject( AnyValue value )
+    public Object valueAsObject( AnyValue value )
     {
         value.writeTo( this );
         return value();

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/BoltStateMachineV1.java

@@ -233,6 +233,12 @@ public void close()
         }
     }
 
+    @Override
+    public String id()
+    {
+        return id;
+    }
+
     @Override
     public String owner()
     {
@@ -351,7 +357,7 @@ private void resetStatementProcessor()
         }
     }
 
-    private static States buildStates()
+    protected States buildStates()
     {
         ConnectedState connected = new ConnectedState();
         ReadyState ready = new ReadyState();
@@ -379,12 +385,12 @@ private static States buildStates()
         return new States( connected, failed );
     }
 
-    private static class States
+    public static class States
     {
         final BoltStateMachineState initial;
         final BoltStateMachineState failed;
 
-        States( BoltStateMachineState initial, BoltStateMachineState failed )
+        public States( BoltStateMachineState initial, BoltStateMachineState failed )
         {
             this.initial = initial;
             this.failed = failed;

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/BoltStateMachineV1SPI.java

@@ -35,7 +35,7 @@
 import org.neo4j.udc.UsageData;
 import org.neo4j.udc.UsageDataKeys;
 
-class BoltStateMachineV1SPI implements BoltStateMachineSPI
+public class BoltStateMachineV1SPI implements BoltStateMachineSPI
 {
     private final BoltConnectionDescriptor connectionDescriptor;
     private final UsageData usageData;
@@ -45,7 +45,7 @@ class BoltStateMachineV1SPI implements BoltStateMachineSPI
     private final String version;
     private final TransactionStateMachineSPI transactionSpi;
 
-    BoltStateMachineV1SPI( BoltConnectionDescriptor connectionDescriptor, UsageData usageData, LogService logging, Authentication authentication,
+    public BoltStateMachineV1SPI( BoltConnectionDescriptor connectionDescriptor, UsageData usageData, LogService logging, Authentication authentication,
             BoltConnectionTracker connectionTracker, TransactionStateMachineSPI transactionStateMachineSPI )
     {
         this.connectionDescriptor = connectionDescriptor;

community/bolt/src/main/java/org/neo4j/bolt/v1/runtime/TransactionStateMachineSPI.java

@@ -52,7 +52,7 @@
 import static java.lang.String.format;
 import static org.neo4j.internal.kernel.api.Transaction.Type.implicit;
 
-class TransactionStateMachineSPI implements org.neo4j.bolt.runtime.TransactionStateMachineSPI
+public class TransactionStateMachineSPI implements org.neo4j.bolt.runtime.TransactionStateMachineSPI
 {
     private static final PropertyContainerLocker locker = new PropertyContainerLocker();
 
@@ -64,7 +64,7 @@ class TransactionStateMachineSPI implements org.neo4j.bolt.runtime.TransactionSt
     private final Duration txAwaitDuration;
     private final Clock clock;
 
-    TransactionStateMachineSPI( GraphDatabaseAPI db, AvailabilityGuard availabilityGuard, Duration txAwaitDuration, Clock clock )
+    public TransactionStateMachineSPI( GraphDatabaseAPI db, AvailabilityGuard availabilityGuard, Duration txAwaitDuration, Clock clock )
     {
         this.db = db;
         this.txBridge = db.getDependencyResolver().resolveDependency( ThreadToStatementContextBridge.class );

community/bolt/src/main/java/org/neo4j/bolt/v2/BoltProtocolV2.java

@@ -22,8 +22,8 @@
 import org.neo4j.bolt.BoltChannel;
 import org.neo4j.bolt.messaging.Neo4jPack;
 import org.neo4j.bolt.runtime.BoltConnectionFactory;
+import org.neo4j.bolt.runtime.BoltStateMachineFactory;
 import org.neo4j.bolt.v1.BoltProtocolV1;
-import org.neo4j.bolt.v1.runtime.BoltStateMachineFactory;
 import org.neo4j.bolt.v2.messaging.Neo4jPackV2;
 import org.neo4j.kernel.impl.logging.LogService;