Closed hantsy closed 11 months ago
I don't think anything can be fixed here, the main problem is that you are using Mono.from
which is canceling the upstream and that is causing all kind of problems. The solution is to use Mono.fromDirect
or use Reactor alternatives of connection/transaction managers.
I did rewrite your code to:
private final R2dbcOperations r2dbcOperations;
private final ReactorConnectionOperations<Connection> connectionOperations;
private final ReactorReactiveTransactionOperations<Connection> transactionOperations;
@Override
public Flux<Customer> findAll() {
var sql = "SELECT * FROM customers ";
return connectionOperations.withConnectionFlux(status -> Flux
.from(status.getConnection().createStatement(sql).execute())
.flatMap(r -> r.map(MAPPING_FUNCTION))
);
}
@Override
public Mono<Customer> findById(UUID id) {
var sql = "SELECT * FROM customers WHERE id=$1 ";
return connectionOperations.withConnectionMono(status -> Mono
.from(status.getConnection().createStatement(sql)
.bind(0, id)
.execute()
)
.flatMap(r -> Mono.from(r.map(MAPPING_FUNCTION)))
.switchIfEmpty(Mono.empty())
);
}
@Override
public Mono<UUID> save(Customer data) {
//var sql = "INSERT INTO customers(name, age, street, city, zip) VALUES (?, ?, ?, ?, ?) RETURNING id ";
var sql = "INSERT INTO customers (name, age, street, city, zip) VALUES ($1, $2, $3, $4, $5)";
return transactionOperations.withTransactionMono(status ->
Mono.just(status.getConnection())
.flatMap(connection -> Mono
.from(connection.createStatement(sql)
.bind(0, data.name())
.bind(1, data.age())
.bind(2, data.address().street())
.bind(3, data.address().street())
.bind(4, data.address().zip())
.returnGeneratedValues("id")
.execute()
)
.flatMap(r -> Mono.from(r.map((row, rowMetadata) -> row.get("id", UUID.class))))
.switchIfEmpty(Mono.empty())
)
);
}
Which makes the test pass even with some remaining Mono.from
. I think we had some hack in v3 and it might have worked.
I think this problem disappears if r2dbc pool is added.
I have used reactive transaction operations in other repository case. Of course it is still working well.
In this example, I only want to use R2dbcOperations, I have to declare there the origin codes is broken as upgrading the new version, in the initial version it is working well.
Not sure if this is something that should be fixed as it's the correct behaviour of R2DBC https://github.com/pgjdbc/r2dbc-postgresql/issues/428#issuecomment-883388654
I don't think it is caused by Mono.from
, in another example to demo the ConnectionFactory
, I also used Mono.from
, it worked, https://github.com/hantsy/micronaut-sandbox/blob/master/data-r2dbc/src/main/java/com/example/CustomerDaoWithConnectionFactory.java
I will reopen the issue and take another look.
The main problem is when the TX block receives the cancel signal, it automatically sends it to the connection block which cancels the connection, so the onCancel
action cannot be executed.
Expected Behavior
I used
R2dbcOperations.withConnection
to build a Repository like this.The complete CustomerDaoWithR2dbcOperations is here.
And the CustomerDaoWithR2dbcOperationsTest failed now.
Actual Behaviour
Steps To Reproduce
Environment Information
Windows 10 pro Java 17
Example Application
https://github.com/hantsy/micronaut-sandbox/tree/master/data-r2dbc
Version
4.0.0