Spring/spring-data-r2dbc
1
0
Fork 0
Code Issues Packages Projects Releases Wiki Activity

#48 - Atomically close connections.

Browse Source 
We now make sure to close connections only once by tracking the cleanup state. Flux.usingWhen/Mono.usingWhen do not ensure atomic cleanup in situations where the subscription completes and then the subscription is terminated.

This behavior has lead to closing a connection multiple times.

Related ticket: reactor/reactor-core#1486
This commit is contained in:
Mark Paluch
2019-01-11 10:25:55 +01:00
parent 2dfb1ff20e
commit 836c5b4ec8
2 changed files with 120 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files

41
src/main/java/org/springframework/data/r2dbc/function/DefaultDatabaseClient.java
View File

@@ -38,6 +38,7 @@ import java.util.LinkedHashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.function.BiFunction;
import java.util.function.Function;
import java.util.function.Supplier;
@@ -122,15 +123,16 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
Assert.notNull(action, "Callback object must not be null");
Mono<Connection> connectionMono = getConnection();
// Create close-suppressing Connection proxy, also preparing returned Statements.
Mono<ConnectionCloseHolder> connectionMono = getConnection()
.map(it -> new ConnectionCloseHolder(it, this::closeConnection));
return Mono.usingWhen(connectionMono, it -> {
Connection connectionToUse = createConnectionProxy(it);
// Create close-suppressing Connection proxy
Connection connectionToUse = createConnectionProxy(it.connection);
return doInConnection(connectionToUse, action);
}, this::closeConnection, this::closeConnection, this::closeConnection) //
}, ConnectionCloseHolder::close, ConnectionCloseHolder::close, ConnectionCloseHolder::close) //
.onErrorMap(R2dbcException.class, ex -> translateException("execute", getSql(action), ex));
}
@@ -149,15 +151,16 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
Assert.notNull(action, "Callback object must not be null");
Mono<Connection> connectionMono = getConnection();
// Create close-suppressing Connection proxy, also preparing returned Statements.
Mono<ConnectionCloseHolder> connectionMono = getConnection()
.map(it -> new ConnectionCloseHolder(it, this::closeConnection));
return Flux.usingWhen(connectionMono, it -> {
Connection connectionToUse = createConnectionProxy(it);
// Create close-suppressing Connection proxy, also preparing returned Statements.
Connection connectionToUse = createConnectionProxy(it.connection);
return doInConnectionMany(connectionToUse, action);
}, this::closeConnection, this::closeConnection, this::closeConnection) //
}, ConnectionCloseHolder::close, ConnectionCloseHolder::close, ConnectionCloseHolder::close) //
.onErrorMap(R2dbcException.class, ex -> translateException("executeMany", getSql(action), ex));
}
@@ -1104,4 +1107,26 @@ class DefaultDatabaseClient implements DatabaseClient, ConnectionAccessor {
}
}
}
/**
* Holder for a connection that makes sure the close action is invoked atomically only once.
*/
@RequiredArgsConstructor
static class ConnectionCloseHolder extends AtomicBoolean {
final Connection connection;
final Function<Connection, Publisher<Void>> closeFunction;
Mono<Void> close() {
return Mono.defer(() -> {
if (compareAndSet(false, true)) {
return Mono.from(closeFunction.apply(connection));
}
return Mono.empty();
});
}
}
}