Byte arrays support

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

@@ -42,7 +42,7 @@ interface Reader
 
     }
 
-    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,6 +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.packstream.ByteArrayIncompatiblePacker;
 import org.neo4j.driver.internal.util.Iterables;
 import org.neo4j.driver.internal.value.InternalValue;
 import org.neo4j.driver.internal.value.ListValue;
@@ -79,10 +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 )
+    public MessageFormat.Writer newWriter( WritableByteChannel ch, boolean byteArraySupportEnabled )
     {
         ChunkedOutput output = new ChunkedOutput( ch );
-        return new Writer( output, output.messageBoundaryHook() );
+        return new Writer( output, output.messageBoundaryHook(), byteArraySupportEnabled );
     }
 
     @Override
@@ -106,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
@@ -223,6 +232,10 @@ private void packValue( Value value ) throws IOException
                     packer.packNull();
                     break;
 
+                case BYTES_TyCon:
+                    packer.pack( value.asByteArray() );
+                    break;
+
                 case STRING_TyCon:
                     packer.pack( value.asString() );
                     break;
@@ -502,8 +515,6 @@ private Value unpackValue() throws IOException
             PackType type = unpacker.peekNextType();
             switch ( type )
             {
-            case BYTES:
-                break;
             case NULL:
                 return value( unpacker.unpackNull() );
             case BOOLEAN:
@@ -512,6 +523,8 @@ private Value unpackValue() throws IOException
                 return value( unpacker.unpackLong() );
             case FLOAT:
                 return value( unpacker.unpackDouble() );
+            case BYTES:
+                return value( unpacker.unpackBytes() );
             case STRING:
                 return value( unpacker.unpackString() );
             case MAP:

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

@@ -40,7 +40,6 @@ public class ChunkedOutput implements PackOutput
     /** Are currently in the middle of writing a chunk? */
     private boolean chunkOpen = false;
 
-
     public ChunkedOutput( WritableByteChannel ch )
     {
         this( 8192, ch );

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

@@ -34,6 +34,8 @@
 
 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
 {
@@ -123,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 )
         {
@@ -139,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;
@@ -255,7 +270,7 @@ private SocketProtocol negotiateProtocol() throws IOException
         {
         case VERSION1:
             logger.debug( "S: [HANDSHAKE] -> 1" );
-            return new SocketProtocolV1( 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

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

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

@@ -18,7 +18,6 @@
  */
 package org.neo4j.driver.internal.net;
 
-import java.io.IOException;
 import java.nio.channels.ByteChannel;
 
 import org.neo4j.driver.internal.messaging.MessageFormat;
@@ -32,15 +31,22 @@ public class SocketProtocolV1 implements SocketProtocol
     private final Reader reader;
     private final Writer writer;
 
-    public SocketProtocolV1( 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 );
-        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/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/PackStream.java

@@ -501,6 +501,29 @@ public double unpackDouble() throws IOException
             throw new Unexpected( "Expected a double, but got: " + toHexString( markerByte ));
         }
 
+        public byte[] unpackBytes() throws IOException
+        {
+            final byte markerByte = in.readByte();
+            switch(markerByte)
+            {
+            case BYTES_8: return unpackRawBytes( unpackUINT8() );
+            case BYTES_16: return unpackRawBytes( unpackUINT16() );
+            case BYTES_32:
+            {
+                long size = unpackUINT32();
+                if ( size <= Integer.MAX_VALUE )
+                {
+                    return unpackRawBytes( (int) size );
+                }
+                else
+                {
+                    throw new Overflow( "BYTES_32 too long for Java" );
+                }
+            }
+            default: throw new Unexpected( "Expected bytes, but got: 0x" + toHexString( markerByte & 0xFF ));
+            }
+        }
+
         public String unpackString() throws IOException
         {
             final byte markerByte = in.readByte();
@@ -512,20 +535,20 @@ public String unpackString() throws IOException
             return new String(unpackUtf8(markerByte), UTF_8);
         }
 
-        public byte[] unpackBytes() throws IOException
+        public byte[] unpackRawBytes() throws IOException
         {
             final byte markerByte = in.readByte();
 
             switch(markerByte)
             {
-                case BYTES_8: return unpackBytes( unpackUINT8() );
-                case BYTES_16: return unpackBytes( unpackUINT16() );
+                case BYTES_8: return unpackRawBytes( unpackUINT8() );
+                case BYTES_16: return unpackRawBytes( unpackUINT16() );
                 case BYTES_32:
                 {
                     long size = unpackUINT32();
                     if ( size <= Integer.MAX_VALUE )
                     {
-                        return unpackBytes( (int) size );
+                        return unpackRawBytes( (int) size );
                     }
                     else
                     {
@@ -558,17 +581,17 @@ private byte[] unpackUtf8(byte markerByte) throws IOException
             final byte markerHighNibble = (byte) (markerByte & 0xF0);
             final byte markerLowNibble = (byte) (markerByte & 0x0F);
 
-            if ( markerHighNibble == TINY_STRING ) { return unpackBytes( markerLowNibble ); }
+            if ( markerHighNibble == TINY_STRING ) { return unpackRawBytes( markerLowNibble ); }
             switch(markerByte)
             {
-                case STRING_8: return unpackBytes( unpackUINT8() );
-                case STRING_16: return unpackBytes( unpackUINT16() );
+                case STRING_8: return unpackRawBytes( unpackUINT8() );
+                case STRING_16: return unpackRawBytes( unpackUINT16() );
                 case STRING_32:
                 {
                     long size = unpackUINT32();
                     if ( size <= Integer.MAX_VALUE )
                     {
-                        return unpackBytes( (int) size );
+                        return unpackRawBytes( (int) size );
                     }
                     else
                     {
@@ -608,7 +631,7 @@ private long unpackUINT32() throws IOException
             return in.readInt() & 0xFFFFFFFFL;
         }
 
-        private byte[] unpackBytes( int size ) throws IOException
+        private byte[] unpackRawBytes(int size ) throws IOException
         {
             byte[] heapBuffer = new byte[size];
             in.readBytes( heapBuffer, 0, heapBuffer.length );
@@ -711,5 +734,4 @@ public UnPackable( String message )
             super( message );
         }
     }
-
 }

driver/src/main/java/org/neo4j/driver/internal/summary/InternalServerInfo.java

@@ -16,6 +16,7 @@
  * See the License for the specific language governing permissions and
  * limitations under the License.
  */
+
 package org.neo4j.driver.internal.summary;
 
 import org.neo4j.driver.internal.net.BoltServerAddress;

driver/src/main/java/org/neo4j/driver/internal/types/InternalTypeSystem.java

@@ -33,6 +33,7 @@
 import static org.neo4j.driver.internal.types.TypeConstructor.NUMBER_TyCon;
 import static org.neo4j.driver.internal.types.TypeConstructor.PATH_TyCon;
 import static org.neo4j.driver.internal.types.TypeConstructor.RELATIONSHIP_TyCon;
+import static org.neo4j.driver.internal.types.TypeConstructor.BYTES_TyCon;
 import static org.neo4j.driver.internal.types.TypeConstructor.STRING_TyCon;
 
 /**
@@ -47,6 +48,7 @@ public class InternalTypeSystem implements TypeSystem
 
     private final TypeRepresentation anyType = constructType( ANY_TyCon );
     private final TypeRepresentation booleanType = constructType( BOOLEAN_TyCon );
+    private final TypeRepresentation bytesType = constructType( BYTES_TyCon );
     private final TypeRepresentation stringType = constructType( STRING_TyCon );
     private final TypeRepresentation numberType = constructType( NUMBER_TyCon );
     private final TypeRepresentation integerType = constructType( INTEGER_TyCon );
@@ -76,6 +78,13 @@ public Type BOOLEAN()
         return booleanType;
     }
 
+    /** the Cypher type BYTES */
+    @Override
+    public Type BYTES()
+    {
+        return bytesType;
+    }
+
     /** the Cypher type STRING */
     @Override
     public Type STRING()

driver/src/main/java/org/neo4j/driver/internal/types/TypeConstructor.java

@@ -44,6 +44,14 @@ public String typeName()
         }
     },
 
+    BYTES_TyCon {
+        @Override
+        public String typeName()
+        {
+            return "BYTES";
+        }
+    },
+
     STRING_TyCon {
         @Override
         public String typeName()

driver/src/main/java/org/neo4j/driver/internal/util/ServerVersion.java

@@ -54,6 +54,12 @@ public static ServerVersion version( Driver driver )
         }
     }
 
+    public static ServerVersion version( Session session )
+    {
+            String versionString = session.run( "RETURN 1" ).consume().server().version();
+            return version( versionString );
+    }
+
     public static ServerVersion version( String server )
     {
         if ( server == null )