Fixed a test

technige · technige · commit c5775cd3f516 · 2016-02-16T21:00:49.000+01:00
diff --git a/driver/src/main/java/org/neo4j/driver/internal/InternalResultCursor.java b/driver/src/main/java/org/neo4j/driver/internal/InternalResultCursor.java
@@ -38,9 +38,11 @@
 import org.neo4j.driver.v1.ResultSummary;
 import org.neo4j.driver.v1.Statement;
 import org.neo4j.driver.v1.StatementType;
+import org.neo4j.driver.v1.Transaction;
 import org.neo4j.driver.v1.UpdateStatistics;
 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.exceptions.NoSuchRecordException;
 
 import static java.lang.String.format;
@@ -51,6 +53,7 @@
 public class InternalResultCursor extends InternalRecordAccessor implements ResultCursor
 {
     private final Connection connection;
+    private final Transaction transaction;
     private final StreamCollector runResponseCollector;
     private final StreamCollector pullAllResponseCollector;
     private final Queue<Record> recordBuffer = new LinkedList<>();
@@ -64,9 +67,10 @@ public class InternalResultCursor extends InternalRecordAccessor implements Resu
     private long limit = -1;
     private boolean done = false;
 
-    public InternalResultCursor( Connection connection, String statement, Map<String, Value> parameters )
+    public InternalResultCursor( Connection connection, Transaction tx, String statement, Map<String, Value> parameters )
     {
         this.connection = connection;
+        this.transaction = tx;
         this.runResponseCollector = newRunResponseCollector();
         this.pullAllResponseCollector = newPullAllResponseCollector( new Statement( statement, parameters ) );
     }
@@ -172,6 +176,22 @@ StreamCollector pullAllResponseCollector()
         return pullAllResponseCollector;
     }
 
+    private void receiveOne()
+    {
+        try
+        {
+            connection.receiveOne();
+        }
+        catch ( Neo4jException ex )
+        {
+            if (transaction != null)
+            {
+                transaction.failure();
+            }
+            throw ex;
+        }
+    }
+
     @Override
     public boolean isOpen()
     {
@@ -203,7 +223,7 @@ public int index( String key )
     public List<String> keys()
     {
         while (keys == null && !done) {
-            connection.receiveOne();
+            receiveOne();
         }
         return keys;
     }
@@ -253,7 +273,7 @@ else if (done)
         {
             while ( recordBuffer.isEmpty() && !done )
             {
-                connection.receiveOne();
+                receiveOne();
             }
             return recordBuffer.isEmpty() && done;
         }
@@ -282,7 +302,7 @@ else if ( done )
         {
             while ( recordBuffer.isEmpty() && !done )
             {
-                connection.receiveOne();
+                receiveOne();
             }
             return next();
         }
@@ -399,7 +419,7 @@ else if ( done )
         {
             while ( recordBuffer.isEmpty() && !done )
             {
-                connection.receiveOne();
+                receiveOne();
             }
             return peek();
         }
diff --git a/driver/src/main/java/org/neo4j/driver/internal/InternalSession.java b/driver/src/main/java/org/neo4j/driver/internal/InternalSession.java
@@ -60,7 +60,7 @@ public InternalSession( Connection connection, Logger logger )
     public ResultCursor run( String statementText, Map<String,Value> statementParameters )
     {
         ensureConnectionIsValid();
-        InternalResultCursor cursor = new InternalResultCursor( connection, statementText, statementParameters );
+        InternalResultCursor cursor = new InternalResultCursor( connection, null, statementText, statementParameters );
         connection.run( statementText, statementParameters, cursor.runResponseCollector() );
         connection.pullAll( cursor.pullAllResponseCollector() );
         connection.sendAll();
diff --git a/driver/src/main/java/org/neo4j/driver/internal/InternalTransaction.java b/driver/src/main/java/org/neo4j/driver/internal/InternalTransaction.java
@@ -125,7 +125,7 @@ public ResultCursor run( String statementText, Map<String,Value> statementParame
 
         try
         {
-            InternalResultCursor cursor = new InternalResultCursor( conn, statementText, statementParameters );
+            InternalResultCursor cursor = new InternalResultCursor( conn, this, statementText, statementParameters );
             conn.run( statementText, statementParameters, cursor.runResponseCollector() );
             conn.pullAll( cursor.pullAllResponseCollector() );
             conn.sendAll();
@@ -158,7 +158,7 @@ public boolean isOpen()
 
     private void ensureNotFailed()
     {
-        if ( state == State.FAILED )
+        if ( state == State.FAILED || state == State.MARKED_FAILED )
         {
             throw new ClientException(
                 "Cannot run more statements in this transaction, because previous statements in the " +
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
@@ -129,7 +129,7 @@ public int receiveAll()
             {
                 reset( StreamCollector.NO_OP );
                 Neo4jException exception = responseHandler.serverFailure();
-                responseHandler.reset();
+                responseHandler.clearError();
                 throw exception;
             }
             return messageCount;
@@ -162,7 +162,7 @@ public void receiveOne()
             {
                 reset( StreamCollector.NO_OP );
                 Neo4jException exception = responseHandler.serverFailure();
-                responseHandler.reset();
+                responseHandler.clearError();
                 throw exception;
             }
         }
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
@@ -72,7 +72,10 @@ public void handleFailureMessage( String code, String message )
                 error = new DatabaseException( code, message );
                 break;
         }
-        collector.done();
+        if ( collector != null )
+        {
+            collector.done();
+        }
     }
 
     @Override
@@ -226,7 +229,7 @@ public Neo4jException serverFailure()
         return error;
     }
 
-    public void reset()
+    public void clearError()
     {
         error = null;
     }
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
@@ -165,8 +165,16 @@ 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 );
+        try
+        {
+            reset( StreamCollector.NO_OP );
+            sync();
+            release.accept( this );
+        }
+        catch (Exception ex)
+        {
+            dispose();
+        }
     }
 
     @Override
diff --git a/driver/src/test/java/org/neo4j/driver/internal/ExtractTest.java b/driver/src/test/java/org/neo4j/driver/internal/ExtractTest.java
@@ -150,7 +150,7 @@ public void testFields() throws Exception
         Connection connection = mock( Connection.class );
         String statement = "<unknown>";
 
-        InternalResultCursor cursor = new InternalResultCursor( connection, statement, ParameterSupport.NO_PARAMETERS );
+        InternalResultCursor cursor = new InternalResultCursor( connection, null, statement, ParameterSupport.NO_PARAMETERS );
         cursor.runResponseCollector().keys( new String[]{"k1"} );
         cursor.runResponseCollector().done();
         cursor.pullAllResponseCollector().record( new Value[]{value( 42 )} );
diff --git a/driver/src/test/java/org/neo4j/driver/internal/InternalResultCursorTest.java b/driver/src/test/java/org/neo4j/driver/internal/InternalResultCursorTest.java
@@ -503,7 +503,7 @@ private ResultCursor createResult( int numberOfRecords )
         Connection connection = mock( Connection.class );
         String statement = "<unknown>";
 
-        InternalResultCursor cursor = new InternalResultCursor( connection, statement, ParameterSupport.NO_PARAMETERS );
+        InternalResultCursor cursor = new InternalResultCursor( connection, null, statement, ParameterSupport.NO_PARAMETERS );
         cursor.runResponseCollector().keys( new String[]{"k1", "k2"} );
         cursor.runResponseCollector().done();
         for ( int i = 1; i <= numberOfRecords; i++ )
diff --git a/driver/src/test/java/org/neo4j/driver/v1/integration/ErrorIT.java b/driver/src/test/java/org/neo4j/driver/v1/integration/ErrorIT.java
@@ -50,7 +50,8 @@ public void shouldThrowHelpfulSyntaxError() throws Throwable
                                  " ^" );
 
         // When
-        session.run( "invalid statement" ).close();
+        ResultCursor result = session.run( "invalid statement" );
+        result.close();
     }
 
     @Test
diff --git a/driver/src/test/java/org/neo4j/driver/v1/integration/SocketClientIT.java b/driver/src/test/java/org/neo4j/driver/v1/integration/SocketClientIT.java
@@ -99,6 +99,5 @@ public void shouldCloseConnectionWhenReceivingProtocolViolationError() throws Ex
 
         assertThat( client.isOpen(), equalTo( false ) );
         verify( handler, times(1) ).protocolViolationErrorOccurred();
-        verify( handler, times(1) ).collectorsWaiting();
     }
 }

Original file line number	Diff line number	Diff line change
`@@ -38,9 +38,11 @@`
`38`	`38`	`import org.neo4j.driver.v1.ResultSummary;`
`39`	`39`	`import org.neo4j.driver.v1.Statement;`
`40`	`40`	`import org.neo4j.driver.v1.StatementType;`
	`41`	`+import org.neo4j.driver.v1.Transaction;`
`41`	`42`	`import org.neo4j.driver.v1.UpdateStatistics;`
`42`	`43`	`import org.neo4j.driver.v1.Value;`
`43`	`44`	`import org.neo4j.driver.v1.exceptions.ClientException;`
	`45`	`+import org.neo4j.driver.v1.exceptions.Neo4jException;`
`44`	`46`	`import org.neo4j.driver.v1.exceptions.NoSuchRecordException;`
`45`	`47`
`46`	`48`	`import static java.lang.String.format;`
`@@ -51,6 +53,7 @@`
`51`	`53`	`public class InternalResultCursor extends InternalRecordAccessor implements ResultCursor`
`52`	`54`	`{`
`53`	`55`	`private final Connection connection;`
	`56`	`+ private final Transaction transaction;`
`54`	`57`	`private final StreamCollector runResponseCollector;`
`55`	`58`	`private final StreamCollector pullAllResponseCollector;`
`56`	`59`	`private final Queue<Record> recordBuffer = new LinkedList<>();`
`@@ -64,9 +67,10 @@ public class InternalResultCursor extends InternalRecordAccessor implements Resu`
`64`	`67`	`private long limit = -1;`
`65`	`68`	`private boolean done = false;`
`66`	`69`
`67`		`- public InternalResultCursor( Connection connection, String statement, Map<String, Value> parameters )`
	`70`	`+ public InternalResultCursor( Connection connection, Transaction tx, String statement, Map<String, Value> parameters )`
`68`	`71`	`{`
`69`	`72`	`this.connection = connection;`
	`73`	`+ this.transaction = tx;`
`70`	`74`	`this.runResponseCollector = newRunResponseCollector();`
`71`	`75`	`this.pullAllResponseCollector = newPullAllResponseCollector( new Statement( statement, parameters ) );`
`72`	`76`	`}`
`@@ -172,6 +176,22 @@ StreamCollector pullAllResponseCollector()`
`172`	`176`	`return pullAllResponseCollector;`
`173`	`177`	`}`
`174`	`178`
	`179`	`+ private void receiveOne()`
	`180`	`+ {`
	`181`	`+ try`
	`182`	`+ {`
	`183`	`+ connection.receiveOne();`
	`184`	`+ }`
	`185`	`+ catch ( Neo4jException ex )`
	`186`	`+ {`
	`187`	`+ if (transaction != null)`
	`188`	`+ {`
	`189`	`+ transaction.failure();`
	`190`	`+ }`
	`191`	`+ throw ex;`
	`192`	`+ }`
	`193`	`+ }`
	`194`	`+`
`175`	`195`	`@Override`
`176`	`196`	`public boolean isOpen()`
`177`	`197`	`{`
`@@ -203,7 +223,7 @@ public int index( String key )`
`203`	`223`	`public List<String> keys()`
`204`	`224`	`{`
`205`	`225`	`while (keys == null && !done) {`
`206`		`- connection.receiveOne();`
	`226`	`+ receiveOne();`
`207`	`227`	`}`
`208`	`228`	`return keys;`
`209`	`229`	`}`
`@@ -253,7 +273,7 @@ else if (done)`
`253`	`273`	`{`
`254`	`274`	`while ( recordBuffer.isEmpty() && !done )`
`255`	`275`	`{`
`256`		`- connection.receiveOne();`
	`276`	`+ receiveOne();`
`257`	`277`	`}`
`258`	`278`	`return recordBuffer.isEmpty() && done;`
`259`	`279`	`}`
`@@ -282,7 +302,7 @@ else if ( done )`
`282`	`302`	`{`
`283`	`303`	`while ( recordBuffer.isEmpty() && !done )`
`284`	`304`	`{`
`285`		`- connection.receiveOne();`
	`305`	`+ receiveOne();`
`286`	`306`	`}`
`287`	`307`	`return next();`
`288`	`308`	`}`
`@@ -399,7 +419,7 @@ else if ( done )`
`399`	`419`	`{`
`400`	`420`	`while ( recordBuffer.isEmpty() && !done )`
`401`	`421`	`{`
`402`		`- connection.receiveOne();`
	`422`	`+ receiveOne();`
`403`	`423`	`}`
`404`	`424`	`return peek();`
`405`	`425`	`}`
Original file line number	Diff line number	Diff line change
`@@ -60,7 +60,7 @@ public InternalSession( Connection connection, Logger logger )`
`60`	`60`	`public ResultCursor run( String statementText, Map<String,Value> statementParameters )`
`61`	`61`	`{`
`62`	`62`	`ensureConnectionIsValid();`
`63`		`- InternalResultCursor cursor = new InternalResultCursor( connection, statementText, statementParameters );`
	`63`	`+ InternalResultCursor cursor = new InternalResultCursor( connection, null, statementText, statementParameters );`
`64`	`64`	`connection.run( statementText, statementParameters, cursor.runResponseCollector() );`
`65`	`65`	`connection.pullAll( cursor.pullAllResponseCollector() );`
`66`	`66`	`connection.sendAll();`
Original file line number	Diff line number	Diff line change
`@@ -125,7 +125,7 @@ public ResultCursor run( String statementText, Map<String,Value> statementParame`
`125`	`125`
`126`	`126`	`try`
`127`	`127`	`{`
`128`		`- InternalResultCursor cursor = new InternalResultCursor( conn, statementText, statementParameters );`
	`128`	`+ InternalResultCursor cursor = new InternalResultCursor( conn, this, statementText, statementParameters );`
`129`	`129`	`conn.run( statementText, statementParameters, cursor.runResponseCollector() );`
`130`	`130`	`conn.pullAll( cursor.pullAllResponseCollector() );`
`131`	`131`	`conn.sendAll();`
`@@ -158,7 +158,7 @@ public boolean isOpen()`
`158`	`158`
`159`	`159`	`private void ensureNotFailed()`
`160`	`160`	`{`
`161`		`- if ( state == State.FAILED )`
	`161`	`+ if ( state == State.FAILED \|\| state == State.MARKED_FAILED )`
`162`	`162`	`{`
`163`	`163`	`throw new ClientException(`
`164`	`164`	`"Cannot run more statements in this transaction, because previous statements in the " +`
Original file line number	Diff line number	Diff line change
`@@ -129,7 +129,7 @@ public int receiveAll()`
`129`	`129`	`{`
`130`	`130`	`reset( StreamCollector.NO_OP );`
`131`	`131`	`Neo4jException exception = responseHandler.serverFailure();`
`132`		`- responseHandler.reset();`
	`132`	`+ responseHandler.clearError();`
`133`	`133`	`throw exception;`
`134`	`134`	`}`
`135`	`135`	`return messageCount;`
`@@ -162,7 +162,7 @@ public void receiveOne()`
`162`	`162`	`{`
`163`	`163`	`reset( StreamCollector.NO_OP );`
`164`	`164`	`Neo4jException exception = responseHandler.serverFailure();`
`165`		`- responseHandler.reset();`
	`165`	`+ responseHandler.clearError();`
`166`	`166`	`throw exception;`
`167`	`167`	`}`
`168`	`168`	`}`
Original file line number	Diff line number	Diff line change
`@@ -72,7 +72,10 @@ public void handleFailureMessage( String code, String message )`
`72`	`72`	`error = new DatabaseException( code, message );`
`73`	`73`	`break;`
`74`	`74`	`}`
`75`		`- collector.done();`
	`75`	`+ if ( collector != null )`
	`76`	`+ {`
	`77`	`+ collector.done();`
	`78`	`+ }`
`76`	`79`	`}`
`77`	`80`
`78`	`81`	`@Override`
`@@ -226,7 +229,7 @@ public Neo4jException serverFailure()`
`226`	`229`	`return error;`
`227`	`230`	`}`
`228`	`231`
`229`		`- public void reset()`
	`232`	`+ public void clearError()`
`230`	`233`	`{`
`231`	`234`	`error = null;`
`232`	`235`	`}`
Original file line number	Diff line number	Diff line change
`@@ -50,7 +50,8 @@ public void shouldThrowHelpfulSyntaxError() throws Throwable`
`50`	`50`	`" ^" );`
`51`	`51`
`52`	`52`	`// When`
`53`		`- session.run( "invalid statement" ).close();`
	`53`	`+ ResultCursor result = session.run( "invalid statement" );`
	`54`	`+ result.close();`
`54`	`55`	`}`
`55`	`56`
`56`	`57`	`@Test`
Original file line number	Diff line number	Diff line change
`@@ -99,6 +99,5 @@ public void shouldCloseConnectionWhenReceivingProtocolViolationError() throws Ex`
`99`	`99`
`100`	`100`	`assertThat( client.isOpen(), equalTo( false ) );`
`101`	`101`	`verify( handler, times(1) ).protocolViolationErrorOccurred();`
`102`		`- verify( handler, times(1) ).collectorsWaiting();`
`103`	`102`	`}`
`104`	`103`	`}`