Remove the two-direction reference between connection and packing protocol.

Background:
A connection holds a socketClient. A socketClient has a protocol which specifies what protocol version the driver should work on.
When a connection is created in a pool, the connection will first be `started` (where the socketClient will negotiate protocol version with the server) and then `initialized` (where the server version will be sent back).

Problem description:
The problem that we need to solve here is that for 3.2+ servers, the protocol the driver is working on should allow bytes, while for server version lower than 3.2.0, the protocol should disallow bytes to be sent to prevent from crashing server connection in a bad way. While the server version is only known after the protocol is created, so after the server version arrived the protocol need either be updated (this commit) or the protocol need to be informed (previous impl).

Solution:
This commit changed the behaviour in connection to update protocol regarding bytes support right after server version is received. By doing in this way we avoid some two-direction reference between connection and protocol.

Author: Zhen

driver/src/main/java/org/neo4j/driver/internal/cluster/RoutingPooledConnection.java

@@ -220,12 +220,6 @@ public BoltServerAddress boltServerAddress()
         return delegate.boltServerAddress();
     }
 
-    @Override
-    public boolean supportsBytes()
-    {
-        return delegate.supportsBytes();
-    }
-
     @Override
     public long lastUsedTimestamp()
     {

driver/src/main/java/org/neo4j/driver/internal/messaging/MessageFormat.java

@@ -18,8 +18,6 @@
  */
 package org.neo4j.driver.internal.messaging;
 
-import org.neo4j.driver.internal.spi.Connection;
-
 import java.io.IOException;
 import java.nio.channels.ReadableByteChannel;
 import java.nio.channels.WritableByteChannel;
@@ -44,9 +42,7 @@ interface Reader
 
     }
 
-    Writer newWriter( WritableByteChannel ch, Connection connection);
-
-    Writer newWriter( WritableByteChannel ch );
+    Writer newWriter( WritableByteChannel ch, boolean byteArraySupportEnabled );
 
     Reader newReader( ReadableByteChannel ch );
 

driver/src/main/java/org/neo4j/driver/internal/messaging/PackStreamMessageFormatV1.java

@@ -37,7 +37,7 @@
 import org.neo4j.driver.internal.packstream.PackOutput;
 import org.neo4j.driver.internal.packstream.PackStream;
 import org.neo4j.driver.internal.packstream.PackType;
-import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.packstream.ByteArrayIncompatiblePacker;
 import org.neo4j.driver.internal.util.Iterables;
 import org.neo4j.driver.internal.value.InternalValue;
 import org.neo4j.driver.internal.value.ListValue;
@@ -80,16 +80,10 @@ public class PackStreamMessageFormatV1 implements MessageFormat
     private static final Map<String,Value> EMPTY_STRING_VALUE_MAP = new HashMap<>( 0 );
 
     @Override
-    public MessageFormat.Writer newWriter( WritableByteChannel ch, Connection connection )
+    public MessageFormat.Writer newWriter( WritableByteChannel ch, boolean byteArraySupportEnabled )
     {
-        ChunkedOutput output = new ChunkedOutput(ch, connection);
-        return new Writer( output, output.messageBoundaryHook() );
-    }
-
-    @Override
-    public MessageFormat.Writer newWriter( WritableByteChannel ch )
-    {
-        return newWriter( ch, null );
+        ChunkedOutput output = new ChunkedOutput( ch );
+        return new Writer( output, output.messageBoundaryHook(), byteArraySupportEnabled );
     }
 
     @Override
@@ -113,11 +107,19 @@ public static class Writer implements MessageFormat.Writer, MessageHandler
         /**
          * @param output interface to write messages to
          * @param onMessageComplete invoked for each message, after it's done writing to the output
+         * @param byteArraySupportEnabled specify if support to pack/write byte array to server
          */
-        public Writer( PackOutput output, Runnable onMessageComplete )
+        public Writer( PackOutput output, Runnable onMessageComplete, boolean byteArraySupportEnabled )
         {
             this.onMessageComplete = onMessageComplete;
-            packer = new PackStream.Packer( output );
+            if( byteArraySupportEnabled )
+            {
+                packer = new PackStream.Packer( output );
+            }
+            else
+            {
+                packer = new ByteArrayIncompatiblePacker( output );
+            }
         }
 
         @Override

driver/src/main/java/org/neo4j/driver/internal/net/ChunkedOutput.java

@@ -23,7 +23,6 @@
 import java.nio.channels.WritableByteChannel;
 
 import org.neo4j.driver.internal.packstream.PackOutput;
-import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.v1.exceptions.ClientException;
 
 import static java.lang.Math.max;
@@ -33,7 +32,6 @@ public class ChunkedOutput implements PackOutput
     public static final short MESSAGE_BOUNDARY = 0;
     public static final int CHUNK_HEADER_SIZE = 2;
 
-    private final Connection connection;
     private final ByteBuffer buffer;
     private final WritableByteChannel channel;
 
@@ -42,14 +40,13 @@ public class ChunkedOutput implements PackOutput
     /** Are currently in the middle of writing a chunk? */
     private boolean chunkOpen = false;
 
-    public ChunkedOutput( WritableByteChannel ch, Connection connection )
+    public ChunkedOutput( WritableByteChannel ch )
     {
-        this(8192, ch, connection);
+        this( 8192, ch );
     }
 
-    public ChunkedOutput( int bufferSize, WritableByteChannel ch, Connection connection )
+    public ChunkedOutput( int bufferSize, WritableByteChannel ch )
     {
-        this.connection = connection;
         buffer = ByteBuffer.allocate(  max( 16, bufferSize ) );
         chunkOpen = false;
         channel = ch;
@@ -124,12 +121,6 @@ public PackOutput writeBytes( byte[] data, int offset, int length ) throws IOExc
         return this;
     }
 
-    @Override
-    public boolean supportsBytes()
-    {
-        return connection != null && connection.supportsBytes();
-    }
-
     private void closeChunkIfOpen()
     {
         if( chunkOpen )

driver/src/main/java/org/neo4j/driver/internal/net/ConcurrencyGuardingConnection.java

@@ -223,10 +223,4 @@ public BoltServerAddress boltServerAddress()
     {
         return delegate.boltServerAddress();
     }
-
-    @Override
-    public boolean supportsBytes()
-    {
-        return delegate.supportsBytes();
-    }
 }

driver/src/main/java/org/neo4j/driver/internal/net/SocketClient.java

@@ -27,14 +27,15 @@
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.MessageFormat;
 import org.neo4j.driver.internal.security.SecurityPlan;
-import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.util.BytePrinter;
 import org.neo4j.driver.v1.Logger;
 import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.exceptions.ServiceUnavailableException;
 
 import static java.lang.String.format;
 import static java.nio.ByteOrder.BIG_ENDIAN;
+import static org.neo4j.driver.internal.util.ServerVersion.v3_2_0;
+import static org.neo4j.driver.internal.util.ServerVersion.version;
 
 public class SocketClient
 {
@@ -44,7 +45,6 @@ public class SocketClient
     private static final int NO_VERSION = 0;
     private static final int[] SUPPORTED_VERSIONS = new int[]{VERSION1, NO_VERSION, NO_VERSION, NO_VERSION};
 
-    private final Connection connection;
     private final BoltServerAddress address;
     private final SecurityPlan securityPlan;
     private final int timeoutMillis;
@@ -56,9 +56,8 @@ public class SocketClient
 
     private ByteChannel channel;
 
-    public SocketClient( Connection connection, BoltServerAddress address, SecurityPlan securityPlan, int timeoutMillis, Logger logger )
+    public SocketClient( BoltServerAddress address, SecurityPlan securityPlan, int timeoutMillis, Logger logger )
     {
-        this.connection = connection;
         this.address = address;
         this.securityPlan = securityPlan;
         this.timeoutMillis = timeoutMillis;
@@ -126,9 +125,7 @@ public void start()
             {
                 setChannel( ChannelFactory.create( address, securityPlan, timeoutMillis, logger ) );
             }
-            protocol = negotiateProtocol();
-            reader = protocol.reader();
-            writer = protocol.writer();
+            setProtocol( negotiateProtocol() );
         }
         catch ( ConnectException e )
         {
@@ -142,6 +139,21 @@ public void start()
         }
     }
 
+    public void updateProtocol( String serverVersion )
+    {
+        if( version( serverVersion ).lessThan( v3_2_0 ) )
+        {
+            setProtocol( SocketProtocolV1.create( channel, false ) );
+        }
+    }
+
+    private void setProtocol( SocketProtocol protocol )
+    {
+        this.protocol = protocol;
+        this.reader = protocol.reader();
+        this.writer = protocol.writer();
+    }
+
     public void send( Queue<Message> messages ) throws IOException
     {
         int messageCount = 0;
@@ -258,7 +270,7 @@ private SocketProtocol negotiateProtocol() throws IOException
         {
         case VERSION1:
             logger.debug( "S: [HANDSHAKE] -> 1" );
-            return new SocketProtocolV1( connection, channel );
+            return SocketProtocolV1.create( channel );
         case NO_VERSION:
             throw new ClientException( "The server does not support any of the protocol versions supported by " +
                                        "this driver. Ensure that you are using driver and server versions that " +

driver/src/main/java/org/neo4j/driver/internal/net/SocketConnection.java

@@ -61,7 +61,7 @@ public class SocketConnection implements Connection
     SocketConnection( BoltServerAddress address, SecurityPlan securityPlan, int timeoutMillis, Logging logging )
     {
         Logger logger = new DelegatingLogger( logging.getLog( LOG_NAME ), String.valueOf( hashCode() ) );
-        this.socket = new SocketClient( this, address, securityPlan, timeoutMillis, logger );
+        this.socket = new SocketClient( address, securityPlan, timeoutMillis, logger );
         this.responseHandler = createResponseHandler( logger );
 
         startSocketClient();
@@ -117,6 +117,7 @@ public void init( String clientName, Map<String,Value> authToken )
         queueMessage( new InitMessage( clientName, authToken ), initCollector );
         sync();
         this.serverInfo = new InternalServerInfo( socket.address(), initCollector.serverVersion() );
+        socket.updateProtocol( serverInfo.version() );
     }
 
     @Override
@@ -167,6 +168,7 @@ public synchronized void flush()
         }
         catch ( IOException e )
         {
+            close();
             throw new ServiceUnavailableException( "Unable to send messages to server: " + e.getMessage(), e );
         }
     }
@@ -303,10 +305,4 @@ public BoltServerAddress boltServerAddress()
     {
         return this.serverInfo.boltServerAddress();
     }
-
-    @Override
-    public boolean supportsBytes()
-    {
-        return this.serverInfo.atLeast("Neo4j", 3, 2);
-    }
 }

driver/src/main/java/org/neo4j/driver/internal/net/SocketProtocolV1.java

@@ -18,30 +18,35 @@
  */
 package org.neo4j.driver.internal.net;
 
-import java.io.IOException;
 import java.nio.channels.ByteChannel;
 
 import org.neo4j.driver.internal.messaging.MessageFormat;
 import org.neo4j.driver.internal.messaging.MessageFormat.Reader;
 import org.neo4j.driver.internal.messaging.MessageFormat.Writer;
 import org.neo4j.driver.internal.messaging.PackStreamMessageFormatV1;
-import org.neo4j.driver.internal.spi.Connection;
 
 public class SocketProtocolV1 implements SocketProtocol
 {
     private final MessageFormat messageFormat;
     private final Reader reader;
     private final Writer writer;
 
-    public SocketProtocolV1( Connection connection, ByteChannel channel ) throws IOException
+    public static SocketProtocol create( ByteChannel channel )
     {
-        messageFormat = new PackStreamMessageFormatV1();
+        /*by default the byte array support is enabled*/
+        return create( channel, true );
+    }
 
-        ChunkedOutput output = new ChunkedOutput(channel, connection);
-        BufferingChunkedInput input = new BufferingChunkedInput( channel );
+    public static SocketProtocol create( ByteChannel channel, boolean byteArraySupportEnabled )
+    {
+        return new SocketProtocolV1( channel, byteArraySupportEnabled );
+    }
 
-        this.writer = new PackStreamMessageFormatV1.Writer( output, output.messageBoundaryHook() );
-        this.reader = new PackStreamMessageFormatV1.Reader( input, input.messageBoundaryHook() );
+    public SocketProtocolV1( ByteChannel channel, boolean byteArraySupportEnabled )
+    {
+        messageFormat = new PackStreamMessageFormatV1();
+        this.writer = messageFormat.newWriter( channel, byteArraySupportEnabled );
+        this.reader = messageFormat.newReader( channel );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/net/pooling/PooledSocketConnection.java

@@ -22,14 +22,12 @@
 
 import org.neo4j.driver.internal.SessionResourcesHandler;
 import org.neo4j.driver.internal.net.BoltServerAddress;
-import org.neo4j.driver.internal.packstream.PackStream;
 import org.neo4j.driver.internal.spi.Collector;
 import org.neo4j.driver.internal.spi.Connection;
 import org.neo4j.driver.internal.spi.PooledConnection;
 import org.neo4j.driver.internal.util.Clock;
 import org.neo4j.driver.internal.util.Consumer;
 import org.neo4j.driver.v1.Value;
-import org.neo4j.driver.v1.exceptions.ClientException;
 import org.neo4j.driver.v1.exceptions.Neo4jException;
 import org.neo4j.driver.v1.summary.ServerInfo;
 
@@ -171,10 +169,6 @@ public void flush()
         {
             delegate.flush();
         }
-        catch ( PackStream.BytesNotSupportedException e )
-        {
-            throw new ClientException("PackStream BYTES are not supported by this server");
-        }
         catch ( RuntimeException e )
         {
             onDelegateException( e );
@@ -251,12 +245,6 @@ public BoltServerAddress boltServerAddress()
         return delegate.boltServerAddress();
     }
 
-    @Override
-    public boolean supportsBytes()
-    {
-        return delegate.supportsBytes();
-    }
-
     @Override
     public void dispose()
     {

driver/src/main/java/org/neo4j/driver/internal/packstream/ByteArrayIncompatiblePacker.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright (c) 2002-2017 "Neo Technology,"
+ * Network Engine for Objects in Lund AB [http://neotechnology.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 org.neo4j.driver.internal.packstream;
+
+import java.io.IOException;
+
+public class ByteArrayIncompatiblePacker extends PackStream.Packer
+{
+    public ByteArrayIncompatiblePacker( PackOutput out )
+    {
+        super( out );
+    }
+
+    public void packBytesHeader( int size ) throws IOException
+    {
+        throw new PackStream.UnPackable( "Packing bytes is not supported " +
+                "as the current server this driver connected to does not support unpack bytes." );
+    }
+}

driver/src/main/java/org/neo4j/driver/internal/packstream/PackOutput.java

@@ -45,7 +45,4 @@ public interface PackOutput
 
     /** Produce an 8-byte IEEE 754 "double format" floating-point number */
     PackOutput writeDouble( double value ) throws IOException;
-
-    /** Return a boolean indicating whether or not this output channel supports the PackStream BYTES type. */
-    boolean supportsBytes();
 }