Handle applications not closing sessions better.

jakewins · jakewins · commit 1fb7d0024e40 · 2016-01-20T15:54:24.000-06:00
- 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.
diff --git a/driver/src/main/java/org/neo4j/driver/internal/InternalSession.java b/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 )
diff --git a/driver/src/main/java/org/neo4j/driver/internal/connector/socket/LoggingResponseHandler.java b/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
diff --git a/driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketConnection.java b/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();
             }
         }
diff --git a/driver/src/main/java/org/neo4j/driver/internal/connector/socket/SocketResponseHandler.java b/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()
     {
 
     }
diff --git a/driver/src/main/java/org/neo4j/driver/internal/messaging/MessageHandler.java b/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;
diff --git a/driver/src/main/java/org/neo4j/driver/internal/messaging/PackStreamMessageFormatV1.java b/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() );
diff --git a/driver/src/main/java/org/neo4j/driver/internal/messaging/ResetMessage.java b/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
diff --git a/driver/src/main/java/org/neo4j/driver/internal/pool/InternalConnectionPool.java b/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(
diff --git a/driver/src/main/java/org/neo4j/driver/internal/pool/PooledConnection.java b/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 );
     }
 
diff --git a/driver/src/main/java/org/neo4j/driver/internal/spi/Connection.java b/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.
      */
diff --git a/driver/src/main/java/org/neo4j/driver/v1/Driver.java b/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" ) );
     }
 
     /**
diff --git a/driver/src/test/java/org/neo4j/driver/internal/InternalSessionTest.java b/driver/src/test/java/org/neo4j/driver/internal/InternalSessionTest.java
diff --git a/driver/src/test/java/org/neo4j/driver/internal/connector/socket/LoggingResponseHandlerTest.java b/driver/src/test/java/org/neo4j/driver/internal/connector/socket/LoggingResponseHandlerTest.java
diff --git a/driver/src/test/java/org/neo4j/driver/internal/messaging/MessageFormatTest.java b/driver/src/test/java/org/neo4j/driver/internal/messaging/MessageFormatTest.java
diff --git a/driver/src/test/java/org/neo4j/driver/internal/util/CaptureStdOut.java b/driver/src/test/java/org/neo4j/driver/internal/util/CaptureStdOut.java
diff --git a/driver/src/test/java/org/neo4j/driver/v1/util/DumpMessage.java b/driver/src/test/java/org/neo4j/driver/v1/util/DumpMessage.java

Original file line number	Diff line number	Diff line change
`@@ -24,7 +24,7 @@`
`24`	`24`	`import org.neo4j.driver.internal.spi.Logger;`
`25`	`25`	`import org.neo4j.driver.v1.Value;`
`26`	`26`
`27`		`-import static org.neo4j.driver.internal.messaging.AckFailureMessage.ACK_FAILURE;`
	`27`	`+import static org.neo4j.driver.internal.messaging.ResetMessage.RESET;`
`28`	`28`	`import static org.neo4j.driver.internal.messaging.DiscardAllMessage.DISCARD_ALL;`
`29`	`29`	`import static org.neo4j.driver.internal.messaging.IgnoredMessage.IGNORED;`
`30`	`30`	`import static org.neo4j.driver.internal.messaging.PullAllMessage.PULL_ALL;`
`@@ -68,10 +68,10 @@ public void handleDiscardAllMessage()`
`68`	`68`	`}`
`69`	`69`
`70`	`70`	`@Override`
`71`		`- public void handleAckFailureMessage()`
	`71`	`+ public void handleResetMessage()`
`72`	`72`	`{`
`73`		`- super.handleAckFailureMessage();`
`74`		`- logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, ACK_FAILURE );`
	`73`	`+ super.handleResetMessage();`
	`74`	`+ logger.debug( DEFAULT_DEBUG_LOGGING_FORMAT, RESET );`
`75`	`75`	`}`
`76`	`76`
`77`	`77`	`@Override`
Original file line number	Diff line number	Diff line change
`@@ -23,7 +23,7 @@`
`23`	`23`	`import java.util.LinkedList;`
`24`	`24`	`import java.util.Map;`
`25`	`25`
`26`		`-import org.neo4j.driver.internal.messaging.AckFailureMessage;`
	`26`	`+import org.neo4j.driver.internal.messaging.ResetMessage;`
`27`	`27`	`import org.neo4j.driver.internal.messaging.InitMessage;`
`28`	`28`	`import org.neo4j.driver.internal.messaging.Message;`
`29`	`29`	`import org.neo4j.driver.internal.messaging.PullAllMessage;`
`@@ -94,6 +94,13 @@ public void pullAll( StreamCollector collector )`
`94`	`94`	`responseHandler.registerResultCollector( messageId, collector );`
`95`	`95`	`}`
`96`	`96`
	`97`	`+ @Override`
	`98`	`+ public void reset( StreamCollector collector )`
	`99`	`+ {`
	`100`	`+ int messageId = queueMessage( ResetMessage.RESET );`
	`101`	`+ responseHandler.registerResultCollector( messageId, collector );`
	`102`	`+ }`
	`103`	`+`
`97`	`104`	`@Override`
`98`	`105`	`public void sync()`
`99`	`106`	`{`
`@@ -111,7 +118,7 @@ public void sync()`
`111`	`118`	`{`
`112`	`119`	`// Its enough to simply add the ack message to the outbound queue, it'll value sent`
`113`	`120`	`// off as the first message the next time we need to sync with the database.`
`114`		`- queueMessage( new AckFailureMessage() );`
	`121`	`+ reset( StreamCollector.NO_OP );`
`115`	`122`	`throw responseHandler.serverFailure();`
`116`	`123`	`}`
`117`	`124`	`}`
Original file line number	Diff line number	Diff line change
`@@ -194,7 +194,7 @@ public void handleDiscardAllMessage()`
`194`	`194`	`}`
`195`	`195`
`196`	`196`	`@Override`
`197`		`- public void handleAckFailureMessage()`
	`197`	`+ public void handleResetMessage()`
`198`	`198`	`{`
`199`	`199`
`200`	`200`	`}`
Original file line number	Diff line number	Diff line change
`@@ -59,7 +59,7 @@`
`59`	`59`	`public class PackStreamMessageFormatV1 implements MessageFormat`
`60`	`60`	`{`
`61`	`61`	`public final static byte MSG_INIT = 0x01;`
`62`		`- public final static byte MSG_ACK_FAILURE = 0x0F;`
	`62`	`+ public final static byte MSG_RESET = 0x0F;`
`63`	`63`	`public final static byte MSG_RUN = 0x10;`
`64`	`64`	`public final static byte MSG_DISCARD_ALL = 0x2F;`
`65`	`65`	`public final static byte MSG_PULL_ALL = 0x3F;`
`@@ -152,9 +152,9 @@ public void handleDiscardAllMessage() throws IOException`
`152`	`152`	`}`
`153`	`153`
`154`	`154`	`@Override`
`155`		`- public void handleAckFailureMessage() throws IOException`
	`155`	`+ public void handleResetMessage() throws IOException`
`156`	`156`	`{`
`157`		`- packer.packStructHeader( 0, MSG_ACK_FAILURE );`
	`157`	`+ packer.packStructHeader( 0, MSG_RESET );`
`158`	`158`	`onMessageComplete.run();`
`159`	`159`	`}`
`160`	`160`
`@@ -429,11 +429,20 @@ public void read( MessageHandler handler ) throws IOException`
`429`	`429`	`case MSG_INIT:`
`430`	`430`	`unpackInitMessage( handler );`
`431`	`431`	`break;`
	`432`	`+ case MSG_RESET:`
	`433`	`+ unpackResetMessage( handler );`
	`434`	`+ break;`
`432`	`435`	`default:`
`433`	`436`	`throw new IOException( "Unknown message type: " + type );`
`434`	`437`	`}`
`435`	`438`	`}`
`436`	`439`
	`440`	`+ private void unpackResetMessage( MessageHandler handler ) throws IOException`
	`441`	`+ {`
	`442`	`+ handler.handleResetMessage();`
	`443`	`+ onMessageComplete.run();`
	`444`	`+ }`
	`445`	`+`
`437`	`446`	`private void unpackInitMessage( MessageHandler handler ) throws IOException`
`438`	`447`	`{`
`439`	`448`	`handler.handleInitMessage( unpacker.unpackString() );`
Original file line number	Diff line number	Diff line change
`@@ -92,7 +92,7 @@ public Connection acquire( URI sessionURI )`
`92`	`92`	`{`
`93`	`93`	`try`
`94`	`94`	`{`
`95`		`- PooledConnection conn = pool( sessionURI ).acquire( 30, TimeUnit.SECONDS );`
	`95`	`+ Connection conn = pool( sessionURI ).acquire( 30, TimeUnit.SECONDS );`
`96`	`96`	`if( conn == null )`
`97`	`97`	`{`
`98`	`98`	`throw new ClientException(`
Original file line number	Diff line number	Diff line change
`@@ -93,6 +93,19 @@ public void pullAll( StreamCollector collector )`
`93`	`93`	`}`
`94`	`94`	`}`
`95`	`95`
	`96`	`+ @Override`
	`97`	`+ public void reset( StreamCollector collector )`
	`98`	`+ {`
	`99`	`+ try`
	`100`	`+ {`
	`101`	`+ delegate.reset( collector );`
	`102`	`+ }`
	`103`	`+ catch ( RuntimeException e )`
	`104`	`+ {`
	`105`	`+ onDelegateException( e );`
	`106`	`+ }`
	`107`	`+ }`
	`108`	`+`
`96`	`109`	`@Override`
`97`	`110`	`public void sync()`
`98`	`111`	`{`
`@@ -109,6 +122,9 @@ public void sync()`
`109`	`122`	`@Override`
`110`	`123`	`public void close()`
`111`	`124`	`{`
	`125`	`+ // In case this session has an open result or transaction or something,`
	`126`	`+ // make sure it's reset to a nice state before we reuse it.`
	`127`	`+ reset( StreamCollector.NO_OP );`
`112`	`128`	`release.accept( this );`
`113`	`129`	`}`
`114`	`130`