Make sure JDBCCursor registers and unregisters with transaction, ensuring it gets closed on transaction commit.

2 files changed, 15 insertions, 5 deletions

diff --git a/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCCursor.java b/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCCursor.java
index 4f31d2c..585b29a 100644
--- a/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCCursor.java
+++ b/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCCursor.java
@@ -28,6 +28,8 @@ import com.amazon.carbonado.FetchException;
 import com.amazon.carbonado.Storable;

 import com.amazon.carbonado.cursor.AbstractCursor;

 

+import com.amazon.carbonado.txn.TransactionScope;

+

 /**

  * Cursor implementation that queries a PreparedStatement.

  *

@@ -35,6 +37,7 @@ import com.amazon.carbonado.cursor.AbstractCursor;
  */

 class JDBCCursor<S extends Storable> extends AbstractCursor<S> {

     private final JDBCStorage<S> mStorage;

+    private final TransactionScope<JDBCTransaction> mScope;

     private final Connection mConnection;

     private final PreparedStatement mStatement;

 

@@ -46,19 +49,23 @@ class JDBCCursor<S extends Storable> extends AbstractCursor<S> {
      * up when this happens by closing statement and connection.

      */

     JDBCCursor(JDBCStorage<S> storage,

+               TransactionScope<JDBCTransaction> scope,

                Connection con,

                PreparedStatement statement)

         throws SQLException

     {

         mStorage = storage;

+        mScope = scope;

         mConnection = con;

         mStatement = statement;

         mResultSet = statement.executeQuery();

+        scope.register(storage.getStorableType(), this);

     }

 

     public void close() throws FetchException {

         if (mResultSet != null) {

             try {

+                mScope.unregister(mStorage.getStorableType(), this);

                 mResultSet.close();

                 mStatement.close();

                 mStorage.mRepository.yieldConnection(mConnection);

diff --git a/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCStorage.java b/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCStorage.java
index cc9bef9..cbd4fc0 100644
--- a/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCStorage.java
+++ b/src/main/java/com/amazon/carbonado/repo/jdbc/JDBCStorage.java
@@ -68,6 +68,7 @@ import com.amazon.carbonado.qe.StandardQuery;
 import com.amazon.carbonado.qe.StandardQueryFactory;

 import com.amazon.carbonado.sequence.SequenceValueProducer;

 import com.amazon.carbonado.spi.TriggerManager;

+import com.amazon.carbonado.txn.TransactionScope;

 import com.amazon.carbonado.util.QuickConstructorGenerator;

 

 /**

@@ -644,7 +645,8 @@ class JDBCStorage<S extends Storable> extends StandardQueryFactory<S>
         }

 

         public Cursor<S> fetch(FilterValues<S> values) throws FetchException {

-            boolean forUpdate = mRepository.localTransactionScope().isForUpdate();

+            TransactionScope<JDBCTransaction> scope = mRepository.localTransactionScope();

+            boolean forUpdate = scope.isForUpdate();

             Connection con = getConnection();

             try {

                 PreparedStatement ps = con.prepareStatement(prepareSelect(values, forUpdate));

@@ -655,7 +657,7 @@ class JDBCStorage<S extends Storable> extends StandardQueryFactory<S>
 

                 try {

                     setParameters(ps, values);

-                    return new JDBCCursor<S>(JDBCStorage.this, con, ps);

+                    return new JDBCCursor<S>(JDBCStorage.this, scope, con, ps);

                 } catch (Exception e) {

                     // in case of exception, close statement

                     try {

@@ -713,7 +715,8 @@ class JDBCStorage<S extends Storable> extends StandardQueryFactory<S>
                 break;

             }

 

-            if (mRepository.localTransactionScope().isForUpdate()) {

+            TransactionScope<JDBCTransaction> scope = mRepository.localTransactionScope();

+            if (scope.isForUpdate()) {

                 select = select.concat(" FOR UPDATE");

             }

 

@@ -733,7 +736,7 @@ class JDBCStorage<S extends Storable> extends StandardQueryFactory<S>
                             switch (option) {

                             case OFFSET_ONLY:

                                 ps.setLong(psOrdinal, from);

-                                Cursor<S> c = new JDBCCursor<S>(JDBCStorage.this, con, ps);

+                                Cursor<S> c = new JDBCCursor<S>(JDBCStorage.this, scope, con, ps);

                                 return new LimitCursor<S>(c, to - from);

                             case LIMIT_AND_OFFSET:

                                 ps.setLong(psOrdinal, to - from);

@@ -755,7 +758,7 @@ class JDBCStorage<S extends Storable> extends StandardQueryFactory<S>
                         ps.setLong(psOrdinal, to);

                     }

 

-                    return new JDBCCursor<S>(JDBCStorage.this, con, ps);

+                    return new JDBCCursor<S>(JDBCStorage.this, scope, con, ps);

                 } catch (Exception e) {

                     // in case of exception, close statement

                     try {