Handle applications not closing sessions better.

- Introduces basic finalizer to the InternalSession object, which closes
  the underlying connection if the session was not closed by the user,
  and prints a warning to stderr.

Author: jakewins

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

driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java

@@ -52,6 +52,12 @@ public interface Connection extends AutoCloseable
      */
     void pullAll( StreamCollector collector );
 
+    /**
+     * Queue a reset action, output will be handed to the collector once the pull starts. This will
+     * close the stream once its completed, allowing another {@link #run(String, java.util.Map, StreamCollector) run}
+     */
+    void reset( StreamCollector collector );
+
     /**
      * Ensure all outstanding actions are carried out on the server.
      */

driver/src/main/java/org/neo4j/driver/v1/Driver.java

@@ -74,11 +74,13 @@ public class Driver implements AutoCloseable
 {
     private final ConnectionPool connections;
     private final URI url;
+    private final Config config;
 
     public Driver( URI url, Config config )
     {
         this.url = url;
         this.connections = new InternalConnectionPool( config );
+        this.config = config;
     }
 
     /**
@@ -88,11 +90,7 @@ public Driver( URI url, Config config )
      */
     public Session session()
     {
-        return new InternalSession( connections.acquire( url ) );
-        // TODO a ConnectionPool per URL
-        // ConnectionPool connections = new InternalConnectionPool( logging, url );
-        // And to value a connection from the pool could be
-        // connections.acquire();
+        return new InternalSession( connections.acquire( url ), config.logging().getLog( "session" ) );
     }
 
     /**