Skip to content
Closed
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
Original file line number Diff line number Diff line change
Expand Up @@ -288,8 +288,12 @@ private SqlClient client() {

@Override
public CompletionStage<Void> beginTransaction() {
if ( transaction != null ) {
throw new IllegalStateException( "Can't begin a new transaction as an active transaction is already associated to this connection" );
}
return connection.begin()
.onSuccess( tx -> LOG.tracef( "Transaction started: %s", tx ) )
.onFailure( v -> LOG.errorf( "Failed to start a transaction: %s", transaction ) )
.toCompletionStage()
.thenAccept( tx -> transaction = tx );
}
Expand All @@ -298,22 +302,28 @@ public CompletionStage<Void> beginTransaction() {
public CompletionStage<Void> commitTransaction() {
return transaction.commit()
.onSuccess( v -> LOG.tracef( "Transaction committed: %s", transaction ) )
.onFailure( v -> LOG.errorf( "Failed to commit transaction: %s", transaction ) )
.toCompletionStage()
.whenComplete( (v, x) -> transaction = null );
.whenComplete( this::afterTransactionEnd );
}

@Override
public CompletionStage<Void> rollbackTransaction() {
return transaction.rollback()
.onFailure( v -> LOG.errorf( "Failed to rollback transaction: %s", transaction ) )
.onSuccess( v -> LOG.tracef( "Transaction rolled back: %s", transaction ) )
.toCompletionStage()
.whenComplete( (v, x) -> transaction = null );
.whenComplete( this::afterTransactionEnd );
}

@Override
public CompletionStage<Void> close() {
if ( transaction != null ) {
throw new IllegalStateException( "Connection being closed with a live transaction associated to it" );
}
return connection.close()
.onSuccess( event -> LOG.tracef( "Connection closed: %s", connection ) )
.onFailure( v -> LOG.errorf( "Failed to close a connection: %s", connection ) )
.toCompletionStage();
}

Expand All @@ -333,6 +343,11 @@ private static <T> T getLastInsertedId(RowSet<Row> rows, Class<T> idClass, Strin
return null;
}

private void afterTransactionEnd(Void v, Throwable x) {
LOG.tracef( "Clearing current transaction instance from connection: %s", transaction );
transaction = null;
}

private static class RowSetResult implements Result {
private final RowSet<Row> rowset;
private final RowIterator<Row> it;
Expand Down