Merge with 1.0

Author: RagnarW

driver/src/main/java/org/neo4j/driver/internal/InternalResultCursor.java

@@ -28,7 +28,7 @@
 import org.neo4j.driver.v1.ResultSummary;
 import org.neo4j.driver.v1.Value;
 import org.neo4j.driver.v1.exceptions.ClientException;
-import org.neo4j.driver.v1.exceptions.NoRecordException;
+import org.neo4j.driver.v1.exceptions.NoSuchRecordException;
 
 import static java.lang.String.format;
 import static java.util.Collections.emptyList;
@@ -100,7 +100,7 @@ public Record record()
         }
         else
         {
-            throw new NoRecordException(
+            throw new NoSuchRecordException(
                 "In order to access the fields of a record in a result, " +
                 "you must first call next() to point the result to the next record in the result stream."
             );
@@ -176,16 +176,38 @@ else if ( records == 0) {
     }
 
     @Override
-    public boolean first()
+    public Record first()
     {
-        long pos = position();
-        return pos < 0 ? next() : pos == 0;
+        if( position() > 0 )
+        {
+            throw new NoSuchRecordException( "Cannot retrieve the first record, because this result cursor has been moved already. " +
+                                             "Please ensure you are not calling `first` multiple times, or are mixing it with calls " +
+                                             "to `next`, `single`, `list` or any other method that changes the position of the cursor." );
+        }
+
+        if( position == 0 )
+        {
+            return record();
+        }
+
+        if( !next() )
+        {
+            throw new NoSuchRecordException( "Cannot retrieve the first record, because this result is empty." );
+        }
+        return record();
     }
 
     @Override
-    public boolean single()
+    public Record single()
     {
-        return first() && atEnd();
+        Record first = first();
+        if( iter.hasNext() )
+        {
+            throw new NoSuchRecordException( "Expected a result with a single record, but this result contains at least one more. " +
+                                             "Ensure your query returns only one record, or use `first` instead of `single` if " +
+                                             "you do not care about the number of records in the result." );
+        }
+        return first;
     }
 
     @Override
@@ -208,7 +230,7 @@ public <T> List<T> list( Function<RecordAccessor, T> mapFunction )
             assertOpen();
             return emptyList();
         }
-        else if ( first() )
+        else if ( position == 0 || ( position == -1 && next() ) )
         {
             List<T> result = new ArrayList<>();
             do

driver/src/main/java/org/neo4j/driver/internal/InternalSession.java

@@ -21,6 +21,7 @@
 import java.util.Map;
 
 import org.neo4j.driver.internal.spi.Connection;
+import org.neo4j.driver.internal.spi.Logger;
 import org.neo4j.driver.internal.summary.ResultBuilder;
 import org.neo4j.driver.internal.types.InternalTypeSystem;
 import org.neo4j.driver.v1.ResultCursor;
@@ -35,6 +36,8 @@ public class InternalSession implements Session
 {
     private final Connection connection;
 
+    private final Logger logger;
+
     /** Called when a transaction object is closed */
     private final Runnable txCleanup = new Runnable()
     {
@@ -48,9 +51,10 @@ public void run()
     private Transaction currentTransaction;
     private boolean isOpen = true;
 
-    public InternalSession( Connection connection )
+    public InternalSession( Connection connection, Logger logger )
     {
         this.connection = connection;
+        this.logger = logger;
     }
 
     @Override
@@ -59,7 +63,6 @@ public ResultCursor run( String statementText, Map<String,Value> statementParame
         ensureConnectionIsValid();
         ResultBuilder resultBuilder = new ResultBuilder( statementText, statementParameters );
         connection.run( statementText, statementParameters, resultBuilder );
-
         connection.pullAll( resultBuilder );
         connection.sync();
         return resultBuilder.build();
@@ -127,6 +130,17 @@ private void ensureConnectionIsValid()
         ensureConnectionIsOpen();
     }
 
+    @Override
+    protected void finalize() throws Throwable
+    {
+        if( isOpen )
+        {
+            logger.error( "Neo4j Session object leaked, please ensure that your application calls the `close` " +
+                          "method on Sessions before disposing of the objects.", null );
+            connection.close();
+        }
+    }
+
     private void ensureNoOpenTransaction()
     {
         if ( currentTransaction != null )

driver/src/main/java/org/neo4j/driver/internal/connector/socket/LoggingResponseHandler.java

@@ -24,7 +24,7 @@
 import org.neo4j.driver.internal.spi.Logger;
 import org.neo4j.driver.v1.Value;
 
-import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;
+import static org.neo4j.driver.internal.messaging.ResetMessage.RESET;
 import static org.neo4j.driver.internal.messaging.DiscardAllMessage.DISCARD_ALL;
 import static org.neo4j.driver.internal.messaging.IgnoredMessage.IGNORED;
 import static org.neo4j.driver.internal.messaging.PullAllMessage.PULL_ALL;
@@ -68,10 +68,10 @@ public void handleDiscardAllMessage()
     }
 
     @Override
-    public void handleAckFailureMessage()
+    public void handleResetMessage()
     {
-        super.handleAckFailureMessage();
-        logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, ACK_FAILURE );
+        super.handleResetMessage();
+        logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, RESET );
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketConnection.java

@@ -23,7 +23,7 @@
 import java.util.LinkedList;
 import java.util.Map;
 
-import org.neo4j.driver.internal.messaging.AckFailureMessage;
+import org.neo4j.driver.internal.messaging.ResetMessage;
 import org.neo4j.driver.internal.messaging.InitMessage;
 import org.neo4j.driver.internal.messaging.Message;
 import org.neo4j.driver.internal.messaging.PullAllMessage;
@@ -94,6 +94,13 @@ public void pullAll( StreamCollector collector )
         responseHandler.registerResultCollector( messageId, collector );
     }
 
+    @Override
+    public void reset( StreamCollector collector )
+    {
+        int messageId = queueMessage( ResetMessage.RESET );
+        responseHandler.registerResultCollector( messageId, collector );
+    }
+
     @Override
     public void sync()
     {
@@ -111,7 +118,7 @@ public void sync()
             {
                 // Its enough to simply add the ack message to the outbound queue, it'll value sent
                 // off as the first message the next time we need to sync with the database.
-                queueMessage( new AckFailureMessage() );
+                reset( StreamCollector.NO_OP );
                 throw responseHandler.serverFailure();
             }
         }

driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketResponseHandler.java

@@ -194,7 +194,7 @@ public void handleDiscardAllMessage()
     }
 
     @Override
-    public void handleAckFailureMessage()
+    public void handleResetMessage()
     {
 
     }

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

@@ -34,7 +34,7 @@ public interface MessageHandler
 
     void handleDiscardAllMessage() throws IOException;
 
-    void handleAckFailureMessage() throws IOException;
+    void handleResetMessage() throws IOException;
 
     // Responses
     void handleSuccessMessage( Map<String,Value> meta ) throws IOException;

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

@@ -59,7 +59,7 @@
 public class PackStreamMessageFormatV1 implements MessageFormat
 {
     public final static byte MSG_INIT = 0x01;
-    public final static byte MSG_ACK_FAILURE = 0x0F;
+    public final static byte MSG_RESET = 0x0F;
     public final static byte MSG_RUN = 0x10;
     public final static byte MSG_DISCARD_ALL = 0x2F;
     public final static byte MSG_PULL_ALL = 0x3F;
@@ -152,9 +152,9 @@ public void handleDiscardAllMessage() throws IOException
         }
 
         @Override
-        public void handleAckFailureMessage() throws IOException
+        public void handleResetMessage() throws IOException
         {
-            packer.packStructHeader( 0, MSG_ACK_FAILURE );
+            packer.packStructHeader( 0, MSG_RESET );
             onMessageComplete.run();
         }
 
@@ -429,11 +429,20 @@ public void read( MessageHandler handler ) throws IOException
             case MSG_INIT:
                 unpackInitMessage( handler );
                 break;
+            case MSG_RESET:
+                unpackResetMessage( handler );
+                break;
             default:
                 throw new IOException( "Unknown message type: " + type );
             }
         }
 
+        private void unpackResetMessage( MessageHandler handler ) throws IOException
+        {
+            handler.handleResetMessage();
+            onMessageComplete.run();
+        }
+
         private void unpackInitMessage( MessageHandler handler ) throws IOException
         {
             handler.handleInitMessage( unpacker.unpackString() );

driver/src/main/java/org/neo4j/driver/internal/messaging/AckFailureMessage.java renamed to driver/src/main/java/org/neo4j/driver/internal/messaging/ResetMessage.java

@@ -21,9 +21,10 @@
 import java.io.IOException;
 
 /**
- * ACK_FAILURE request message
+ * RESET request message
  * <p>
- * Sent by clients to acknowledge receipt of failures sent by the server. This is required to
+ * Sent by clients to reset a session to a clean state - closing any open transaction or result streams.
+ * This also acknowledges receipt of failures sent by the server. This is required to
  * allow optimistic sending of multiple messages before responses have been received - pipelining.
  * <p>
  * When something goes wrong, we want the server to stop processing our already sent messages,
@@ -33,20 +34,20 @@
  * This message acts as a barrier after an error, informing the server that we've seen the error
  * message, and that messages that follow this one are safe to execute.
  */
-public class AckFailureMessage implements Message
+public class ResetMessage implements Message
 {
-    public static final AckFailureMessage ACK_FAILURE = new AckFailureMessage();
+    public static final ResetMessage RESET = new ResetMessage();
 
     @Override
     public void dispatch( MessageHandler handler ) throws IOException
     {
-        handler.handleAckFailureMessage();
+        handler.handleResetMessage();
     }
 
     @Override
     public String toString()
     {
-        return "[ACK_FAILURE]";
+        return "[RESET]";
     }
 
     @Override

driver/src/main/java/org/neo4j/driver/internal/pool/InternalConnectionPool.java

@@ -92,7 +92,7 @@ public Connection acquire( URI sessionURI )
     {
         try
         {
-            PooledConnection conn = pool( sessionURI ).acquire( 30, TimeUnit.SECONDS );
+            Connection conn = pool( sessionURI ).acquire( 30, TimeUnit.SECONDS );
             if( conn == null )
             {
                 throw new ClientException(

driver/src/main/java/org/neo4j/driver/internal/pool/PooledConnection.java

@@ -93,6 +93,19 @@ public void pullAll( StreamCollector collector )
         }
     }
 
+    @Override
+    public void reset( StreamCollector collector )
+    {
+        try
+        {
+            delegate.reset( collector );
+        }
+        catch ( RuntimeException e )
+        {
+            onDelegateException( e );
+        }
+    }
+
     @Override
     public void sync()
     {
@@ -109,6 +122,9 @@ public void sync()
     @Override
     public void close()
     {
+        // In case this session has an open result or transaction or something,
+        // make sure it's reset to a nice state before we reuse it.
+        reset( StreamCollector.NO_OP );
         release.accept( this );
     }