From 988b31b33a8ec2616cf5db31215ad0e1037a1d31 Mon Sep 17 00:00:00 2001 From: Christoph Strobl Date: Wed, 8 May 2019 20:01:06 +0200 Subject: [PATCH] #107 - Polishing. Extract method references for better readability. Add missing tests and update documentation. Add delay for transactional MySql tests to avoid failures due to potentially delayed transaction id storage within the database. Original Pull Request: #107 --- .../reference/r2dbc-transactions.adoc | 76 +++++--- .../ConnectionFactoryTransactionManager.java | 2 +- .../ConnectionFactoryUtils.java | 114 ++++++----- .../connectionfactory/ConnectionHandle.java | 8 +- .../connectionfactory/ConnectionHolder.java | 15 +- .../connectionfactory/ConnectionProxy.java | 6 +- .../DelegatingConnectionFactory.java | 23 ++- .../R2dbcTransactionObjectSupport.java | 5 +- .../SimpleConnectionHandle.java | 8 +- ...ransactionAwareConnectionFactoryProxy.java | 75 +++++--- ...ctionalDatabaseClientIntegrationTests.java | 181 +++++++++++++++--- ...ctionalDatabaseClientIntegrationTests.java | 33 +++- ...ionFactoryTransactionManagerUnitTests.java | 2 +- .../ConnectionFactoryUtilsUnitTests.java | 26 ++- .../DelegatingConnectionFactoryUnitTests.java | 2 +- ...nAwareConnectionFactoryProxyUnitTests.java | 118 ++++++++++-- src/test/resources/logback.xml | 2 +- 17 files changed, 514 insertions(+), 182 deletions(-) diff --git a/src/main/asciidoc/reference/r2dbc-transactions.adoc b/src/main/asciidoc/reference/r2dbc-transactions.adoc index 8917eb8..0f0fdba 100644 --- a/src/main/asciidoc/reference/r2dbc-transactions.adoc +++ b/src/main/asciidoc/reference/r2dbc-transactions.adoc @@ -4,7 +4,8 @@ A common pattern when using relational databases is grouping multiple queries within a unit of work that is guarded by a transaction. Relational databases typically associate a transaction with a single transport connection. Using different connections hence results in utilizing different transactions. -Spring Data R2DBC includes transaction-awareness in `DatabaseClient` that allows you to group multiple statements within the same transaction using https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction[Spring's Transaction Management]. +Spring Data R2DBC includes transaction-awareness in `DatabaseClient` that allows you to group multiple statements within +the same transaction using https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction[Spring's Transaction Management]. Spring Data R2DBC provides a implementation for `ReactiveTransactionManager` with `ConnectionFactoryTransactionManager`. See <> for further details. @@ -12,54 +13,75 @@ See <> for further detail ==== [source,java] ---- -ConnectionFactoryTransactionManager tm = new ConnectionFactoryTransactionManager(connectionFactory); -TransactionalOperator operator = TransactionalOperator.create(tm); -DatabaseClient db = DatabaseClient.create(connectionFactory); +ReactiveTransactionManager tm = new ConnectionFactoryTransactionManager(connectionFactory); +TransactionalOperator operator = TransactionalOperator.create(tm); <1> -Mono atomicOperation = db.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") - .bind("id", "joe") - .bind("name", "Joe") - .bind("age", 34) - .fetch().rowsUpdated() - .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") - .bind("id", "joe") - .bind("name", "Joe") - .fetch().rowsUpdated()) - .then() - .as(operator::transactional); +DatabaseClient client = DatabaseClient.create(connectionFactory); + +Mono atomicOperation = client.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") + .bind("id", "joe") + .bind("name", "Joe") + .bind("age", 34) + .fetch().rowsUpdated() + .then(client.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") + .bind("id", "joe") + .bind("name", "Joe") + .fetch().rowsUpdated()) + .then() + .as(operator::transactional); <2> }); ---- +<1> Associate the `TransactionalOperator` with the `ReactiveTransactionManager`. +<2> Bind the operation to the `TransactionalOperator`. ==== -https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction-declarative[Spring's declarative Transaction Management] is a less invasive, annotation-based approach to transaction demarcation. +https://docs.spring.io/spring/docs/current/spring-framework-reference/data-access.html#transaction-declarative[Spring's declarative Transaction Management] +is a less invasive, annotation-based approach to transaction demarcation. .Declarative Transaction Management ==== [source,java] ---- -class MyService { +@Configuration +@EnableTransactionManagement <1> +class Config extends AbstractR2dbcConfiguration { - private final DatabaseClient db; - - MyService(DatabaseClient db) { - this.db = db; + @Override + public ConnectionFactory connectionFactory() { + return // ... } + @Bean + ReactiveTransactionManager txMgr(ConnectionFactory connectionFactory) { <2> + return new ConnectionFactoryTransactionManager(connectionFactory); + } +} + +@Service +class MyService { + + private final DatabaseClient client; + + MyService(DatabaseClient client) { + this.client = client; + } @Transactional public Mono insertPerson() { - return db.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") + return client.execute().sql("INSERT INTO person (id, name, age) VALUES(:id, :name, :age)") .bind("id", "joe") .bind("name", "Joe") .bind("age", 34) .fetch().rowsUpdated() - .then(db.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") - .bind("id", "joe") - .bind("name", "Joe") - .fetch().rowsUpdated()) - .then(); + .then(client.execute().sql("INSERT INTO contacts (id, name) VALUES(:id, :name)") + .bind("id", "joe") + .bind("name", "Joe") + .fetch().rowsUpdated()) + .then(); } } ---- +<1> Enable declarative transaction management. +<2> Provide a `ReactiveTransactionManager` implementation to back reactive tansaction features. ==== diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java index a5a0ea1..4bb0caf 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManager.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java index c017222..f54db09 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtils.java @@ -17,14 +17,9 @@ package org.springframework.data.r2dbc.function.connectionfactory; import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; -import reactor.core.publisher.Mono; -import reactor.util.function.Tuple2; -import reactor.util.function.Tuples; - import org.apache.commons.logging.Log; import org.apache.commons.logging.LogFactory; import org.reactivestreams.Publisher; - import org.springframework.core.Ordered; import org.springframework.dao.DataAccessResourceFailureException; import org.springframework.lang.Nullable; @@ -32,6 +27,9 @@ import org.springframework.transaction.NoTransactionException; import org.springframework.transaction.reactive.TransactionSynchronization; import org.springframework.transaction.reactive.TransactionSynchronizationManager; import org.springframework.util.Assert; +import reactor.core.publisher.Mono; +import reactor.util.function.Tuple2; +import reactor.util.function.Tuples; /** * Helper class that provides static methods for obtaining R2DBC Connections from a @@ -41,6 +39,7 @@ import org.springframework.util.Assert; * objects. Can also be used directly in application code. * * @author Mark Paluch + * @author Christoph Strobl */ public abstract class ConnectionFactoryUtils { @@ -52,7 +51,6 @@ public abstract class ConnectionFactoryUtils { private static final Log logger = LogFactory.getLog(ConnectionFactoryUtils.class); private ConnectionFactoryUtils() { - } /** @@ -93,14 +91,20 @@ public abstract class ConnectionFactoryUtils { if (conHolder != null && (conHolder.hasConnection() || conHolder.isSynchronizedWithTransaction())) { conHolder.requested(); if (!conHolder.hasConnection()) { - logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); + + if (logger.isDebugEnabled()) { + logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); + } return fetchConnection(connectionFactory).doOnNext(conHolder::setConnection); } return Mono.just(conHolder.getConnection()); } // Else we either got no holder or an empty thread-bound holder here. - logger.debug("Fetching R2DBC Connection from ConnectionFactory"); + if (logger.isDebugEnabled()) { + logger.debug("Fetching R2DBC Connection from ConnectionFactory"); + } + Mono con = fetchConnection(connectionFactory); if (synchronizationManager.isSynchronizationActive()) { @@ -124,11 +128,9 @@ public abstract class ConnectionFactoryUtils { if (holderToUse != conHolder) { synchronizationManager.bindResource(connectionFactory, holderToUse); } - }).onErrorResume(e -> { // Unexpected exception from external delegation call -> close Connection and rethrow. return releaseConnection(it, connectionFactory).then(Mono.error(e)); - }); }); } @@ -158,7 +160,9 @@ public abstract class ConnectionFactoryUtils { if (synchronization.isSynchronizationActive()) { - logger.debug("Registering transaction synchronization for R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Registering transaction synchronization for R2DBC Connection"); + } TransactionResources txContext = synchronization.getCurrentTransaction(); ConnectionFactory resource = txContext.getResource(ConnectionFactory.class); @@ -166,7 +170,9 @@ public abstract class ConnectionFactoryUtils { Mono> attachNewConnection = Mono .defer(() -> Mono.from(connectionFactory.create()).map(it -> { - logger.debug("Fetching new R2DBC Connection from ConnectionFactory"); + if (logger.isDebugEnabled()) { + logger.debug("Fetching new R2DBC Connection from ConnectionFactory"); + } SingletonConnectionFactory s = new SingletonConnectionFactory(connectionFactory.getMetadata(), it); txContext.registerResource(ConnectionFactory.class, s); @@ -174,14 +180,9 @@ public abstract class ConnectionFactoryUtils { return Tuples.of(it, connectionFactory); })); - return Mono.justOrEmpty(resource).flatMap(factory -> { - - logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); - - return Mono.from(factory.create()) - .map(connection -> Tuples. of(connection, factory)); - - }).switchIfEmpty(attachNewConnection); + return Mono.justOrEmpty(resource) + .flatMap(ConnectionFactoryUtils::createConnection) + .switchIfEmpty(attachNewConnection); } return Mono.empty(); @@ -199,7 +200,7 @@ public abstract class ConnectionFactoryUtils { private static Mono fetchConnection(ConnectionFactory connectionFactory) { Publisher con = connectionFactory.create(); - if (con == null) { + if (con == null) { // TODO: seriously why would it do that? throw new IllegalStateException("ConnectionFactory returned null from getConnection(): " + connectionFactory); } return Mono.from(con); @@ -211,7 +212,7 @@ public abstract class ConnectionFactoryUtils { * * @param con the {@link io.r2dbc.spi.Connection} to close if necessary. * @param connectionFactory the {@link ConnectionFactory} that the Connection was obtained from (may be - * {@literal null}). + * {@literal null}). * @see #getConnection */ public static Mono releaseConnection(@Nullable io.r2dbc.spi.Connection con, @@ -227,7 +228,7 @@ public abstract class ConnectionFactoryUtils { * * @param con the {@link io.r2dbc.spi.Connection} to close if necessary. * @param connectionFactory the {@link ConnectionFactory} that the Connection was obtained from (may be - * {@literal null}). + * {@literal null}). * @see #doGetConnection */ public static Mono doReleaseConnection(@Nullable io.r2dbc.spi.Connection con, @@ -247,12 +248,16 @@ public abstract class ConnectionFactoryUtils { SingletonConnectionFactory factory = (SingletonConnectionFactory) connectionFactory; - logger.debug("Releasing R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Releasing R2DBC Connection"); + } return factory.close(con); } - logger.debug("Closing R2DBC Connection"); + if (logger.isDebugEnabled()) { + logger.debug("Closing R2DBC Connection"); + } return Mono.from(con.close()); }); @@ -267,6 +272,7 @@ public abstract class ConnectionFactoryUtils { * @throws DataAccessResourceFailureException if the attempt to get a {@link io.r2dbc.spi.Connection} failed */ public static Mono closeConnection(Connection connection, ConnectionFactory connectionFactory) { + return doCloseConnection(connection, connectionFactory) .onErrorMap(e -> new DataAccessResourceFailureException("Failed to obtain R2DBC Connection", e)); } @@ -295,10 +301,10 @@ public abstract class ConnectionFactoryUtils { * Obtain the currently {@link ReactiveTransactionSynchronization} from the current subscriber * {@link reactor.util.context.Context}. * + * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the + * current subscription. * @see Mono#subscriberContext() * @see ReactiveTransactionSynchronization - * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the - * current subscription. */ public static Mono currentReactiveTransactionSynchronization() { @@ -312,10 +318,10 @@ public abstract class ConnectionFactoryUtils { * Obtain the currently active {@link ReactiveTransactionSynchronization} from the current subscriber * {@link reactor.util.context.Context}. * + * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the + * current subscription. * @see Mono#subscriberContext() * @see ReactiveTransactionSynchronization - * @throws NoTransactionException if no active {@link ReactiveTransactionSynchronization} is associated with the - * current subscription. */ public static Mono currentActiveReactiveTransactionSynchronization() { @@ -341,16 +347,8 @@ public abstract class ConnectionFactoryUtils { return true; } return false; - }).map(it -> connectionFactory).onErrorResume(NoTransactionException.class, e -> { - - return currentActiveReactiveTransactionSynchronization().map(synchronization -> { - - TransactionResources currentSynchronization = synchronization.getCurrentTransaction(); - return currentSynchronization.getResource(ConnectionFactory.class); - }).switchIfEmpty(Mono.error(new DataAccessResourceFailureException( - "Cannot extract ConnectionFactory from current TransactionContext!"))); - - }); + }).map(it -> connectionFactory) // + .onErrorResume(NoTransactionException.class, ConnectionFactoryUtils::obtainDefaultConnectionFactory); } /** @@ -359,7 +357,7 @@ public abstract class ConnectionFactoryUtils { * * @param conHolder the {@link ConnectionHolder} for the held Connection (potentially a proxy) * @param passedInCon the {@link Connection} passed-in by the user (potentially a target {@link Connection} without - * proxy) + * proxy) * @return whether the given Connections are equal * @see #getTargetConnection */ @@ -402,6 +400,7 @@ public abstract class ConnectionFactoryUtils { * @see #CONNECTION_SYNCHRONIZATION_ORDER */ private static int getConnectionSynchronizationOrder(ConnectionFactory connectionFactory) { + int order = CONNECTION_SYNCHRONIZATION_ORDER; ConnectionFactory current = connectionFactory; while (current instanceof DelegatingConnectionFactory) { @@ -411,6 +410,37 @@ public abstract class ConnectionFactoryUtils { return order; } + /** + * @param e + * @return an {@link Mono#error(Throwable) error} if not transaction present. + */ + private static Mono obtainDefaultConnectionFactory(NoTransactionException e) { + + return currentActiveReactiveTransactionSynchronization().map(synchronization -> { + + TransactionResources currentSynchronization = synchronization.getCurrentTransaction(); + return currentSynchronization.getResource(ConnectionFactory.class); + }).switchIfEmpty(Mono.error(new DataAccessResourceFailureException( + "Cannot extract ConnectionFactory from current TransactionContext!"))); + } + + /** + * Create a {@link Connection} via the given {@link ConnectionFactory#create() factory} and return a {@link Tuple2} associating the + * {@link Connection} with its creating {@link ConnectionFactory}. + * + * @param factory must not be {@literal null}. + * @return never {@literal null} + */ + private static Mono> createConnection(ConnectionFactory factory) { + + if (logger.isDebugEnabled()) { + logger.debug("Fetching resumed R2DBC Connection from ConnectionFactory"); + } + + return Mono.from(factory.create()) + .map(connection -> Tuples.of(connection, factory)); + } + /** * Callback for resource cleanup at the end of a non-native R2DBC transaction. */ @@ -465,7 +495,6 @@ public abstract class ConnectionFactoryUtils { }).then(); } return Mono.empty(); - } @Override @@ -510,17 +539,14 @@ public abstract class ConnectionFactoryUtils { // Reset the ConnectionHolder: It might remain bound to the context. this.connectionHolder.setConnection(null); }); - } return Mono.empty(); }); - } this.connectionHolder.reset(); return Mono.empty(); } } - } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java index 155a79c..ed5a4e6 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHandle.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -38,8 +38,8 @@ public interface ConnectionHandle { *

* The default implementation is empty, assuming that the lifecycle of the connection is managed externally. * - * @param con the R2DBC Connection to release + * @param connection the R2DBC Connection to release */ - default void releaseConnection(Connection con) {} - + default void releaseConnection(Connection connection) { + } } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java index fcc3f3b..f84f6c7 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionHolder.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -17,7 +17,6 @@ package org.springframework.data.r2dbc.function.connectionfactory; import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; - import org.springframework.lang.Nullable; import org.springframework.transaction.support.ResourceHolderSupport; import org.springframework.util.Assert; @@ -30,6 +29,8 @@ import org.springframework.util.Assert; *

* Note: This is an SPI class, not intended to be used by applications. * + * @author Mark Paluch + * @author Christoph Strobl * @see ConnectionFactoryTransactionManager * @see ConnectionFactoryUtils */ @@ -39,7 +40,7 @@ public class ConnectionHolder extends ResourceHolderSupport { @Nullable private Connection currentConnection; - private boolean transactionActive = false; + private boolean transactionActive; /** * Create a new ConnectionHolder for the given R2DBC {@link Connection}, wrapping it with a @@ -50,7 +51,7 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see #ConnectionHolder(Connection, boolean) */ public ConnectionHolder(Connection connection) { - this.connectionHandle = new SimpleConnectionHandle(connection); + this(connection, false); } /** @@ -62,7 +63,8 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see SimpleConnectionHandle */ public ConnectionHolder(Connection connection, boolean transactionActive) { - this(connection); + + this.connectionHandle = new SimpleConnectionHandle(connection); this.transactionActive = transactionActive; } @@ -127,6 +129,7 @@ public class ConnectionHolder extends ResourceHolderSupport { * @see #released() */ public Connection getConnection() { + Assert.notNull(this.connectionHandle, "Active Connection is required"); if (this.currentConnection == null) { this.currentConnection = this.connectionHandle.getConnection(); @@ -152,7 +155,7 @@ public class ConnectionHolder extends ResourceHolderSupport { } } - /* + /* * (non-Javadoc) * @see org.springframework.transaction.support.ResourceHolderSupport#clear() */ diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java index 37f600c..38bd461 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionProxy.java @@ -1,5 +1,5 @@ /* - * Copyright 2018-2019 the original author or authors. + * Copyright 2019 the original author or authors. * * Licensed under the Apache License, Version 2.0 (the "License"); * you may not use this file except in compliance with the License. @@ -19,12 +19,13 @@ import io.r2dbc.spi.Connection; import io.r2dbc.spi.Wrapped; /** - * Subinterface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target + * Sub interface of {@link Connection} to be implemented by Connection proxies. Allows access to the underlying target * Connection. *

* This interface can be checked when there is a need to cast to a native R2DBC {@link Connection}. * * @author Mark Paluch + * @author Christoph Strobl */ public interface ConnectionProxy extends Connection, Wrapped { @@ -34,6 +35,7 @@ public interface ConnectionProxy extends Connection, Wrapped { * This will typically be the native driver {@link Connection} or a wrapper from a connection pool. * * @return the underlying Connection (never {@literal null}) + * @throws IllegalStateException in case the connection has already been closed. */ Connection getTargetConnection(); } diff --git a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java index f28b2c8..3b57769 100644 --- a/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java +++ b/src/main/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactory.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -19,7 +19,6 @@ import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; import io.r2dbc.spi.ConnectionFactoryMetadata; import io.r2dbc.spi.Wrapped; - import org.reactivestreams.Publisher; import org.springframework.lang.Nullable; import org.springframework.util.Assert; @@ -43,7 +42,7 @@ public class DelegatingConnectionFactory implements ConnectionFactory, Wrapped getTransactionAwareConnectionProxy(ConnectionFactory targetConnectionFactory) { + return ConnectionFactoryUtils.getConnection(targetConnectionFactory).map(TransactionAwareConnectionFactoryProxy::proxyConnection); + } - return ConnectionFactoryUtils.getConnection(targetConnectionFactory).map(tuple -> { - return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(), - new Class[] { ConnectionProxy.class }, - new TransactionAwareInvocationHandler(tuple.getT1(), targetConnectionFactory)); - }); + private static Connection proxyConnection(Tuple2 connectionConnectionFactoryTuple) { + + return (Connection) Proxy.newProxyInstance(ConnectionProxy.class.getClassLoader(), + new Class[]{ConnectionProxy.class}, + new TransactionAwareInvocationHandler(connectionConnectionFactoryTuple.getT1(), connectionConnectionFactoryTuple.getT2())); } /** * Invocation handler that delegates close calls on R2DBC Connections to {@link ConnectionFactoryUtils} for being * aware of context-bound transactions. */ - private class TransactionAwareInvocationHandler implements InvocationHandler { + private static class TransactionAwareInvocationHandler implements InvocationHandler { private final Connection connection; @@ -116,11 +123,12 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection private boolean closed = false; TransactionAwareInvocationHandler(Connection connection, ConnectionFactory targetConnectionFactory) { + this.connection = connection; this.targetConnectionFactory = targetConnectionFactory; } - /* + /* * (non-Javadoc) * @see java.lang.reflect.InvocationHandler#invoke(java.lang.Object, java.lang.reflect.Method, java.lang.Object[]) */ @@ -128,23 +136,24 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection @Nullable public Object invoke(Object proxy, Method method, Object[] args) throws Throwable { + if (ReflectionUtils.isObjectMethod(method)) { + + if (ReflectionUtils.isToStringMethod(method)) { + return proxyToString(proxy); + } + + if (ReflectionUtils.isEqualsMethod(method)) { + return (proxy == args[0]); + } + + if (ReflectionUtils.isHashCodeMethod(method)) { + return System.identityHashCode(proxy); + } + } + // Invocation on ConnectionProxy interface coming in... switch (method.getName()) { - case "equals": - // Only considered as equal when proxies are identical. - return (proxy == args[0]); - case "hashCode": - // Use hashCode of Connection proxy. - return System.identityHashCode(proxy); - case "toString": - // Allow for differentiating between the proxy and the raw Connection. - StringBuilder sb = new StringBuilder("Transaction-aware proxy for target Connection "); - if (this.connection != null) { - sb.append("[").append(this.connection.toString()).append("]"); - } else { - sb.append(" from ConnectionFactory [").append(this.targetConnectionFactory).append("]"); - } - return sb.toString(); + case "unwrap": return this.connection; case "close": @@ -171,5 +180,17 @@ public class TransactionAwareConnectionFactoryProxy extends DelegatingConnection throw ex.getTargetException(); } } + + private String proxyToString(@Nullable Object proxy) { + + // Allow for differentiating between the proxy and the raw Connection. + StringBuilder sb = new StringBuilder("Transaction-aware proxy for target Connection "); + if (this.connection != null) { + sb.append("[").append(this.connection.toString()).append("]"); + } else { + sb.append(" from ConnectionFactory [").append(this.targetConnectionFactory).append("]"); + } + return sb.toString(); + } } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java index 24a84cc..d47d438 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/AbstractTransactionalDatabaseClientIntegrationTests.java @@ -17,33 +17,41 @@ package org.springframework.data.r2dbc.function; import static org.assertj.core.api.Assertions.*; -import io.r2dbc.spi.ConnectionFactory; -import reactor.core.publisher.Flux; -import reactor.core.publisher.Mono; -import reactor.test.StepVerifier; - +import javax.sql.DataSource; import java.util.ArrayList; import java.util.List; import java.util.Queue; import java.util.concurrent.ArrayBlockingQueue; -import javax.sql.DataSource; - +import io.r2dbc.spi.ConnectionFactory; +import org.junit.After; import org.junit.Before; import org.junit.Test; - +import org.springframework.beans.factory.annotation.Autowired; +import org.springframework.context.annotation.AnnotationConfigApplicationContext; +import org.springframework.context.annotation.Bean; +import org.springframework.context.annotation.Configuration; +import org.springframework.context.support.GenericApplicationContext; import org.springframework.dao.DataAccessException; +import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration; import org.springframework.data.r2dbc.function.connectionfactory.ConnectionFactoryTransactionManager; import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport; import org.springframework.jdbc.core.JdbcTemplate; import org.springframework.transaction.NoTransactionException; +import org.springframework.transaction.ReactiveTransactionManager; +import org.springframework.transaction.annotation.EnableTransactionManagement; +import org.springframework.transaction.annotation.Transactional; import org.springframework.transaction.reactive.TransactionalOperator; import org.springframework.transaction.support.DefaultTransactionDefinition; +import reactor.core.publisher.Flux; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * Abstract base class for integration tests for {@link TransactionalDatabaseClient}. * * @author Mark Paluch + * @author Christoph Strobl */ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extends R2dbcIntegrationTestSupport { @@ -51,19 +59,35 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend private JdbcTemplate jdbc; + AnnotationConfigApplicationContext context; + TransactionalService service; + @Before public void before() { connectionFactory = createConnectionFactory(); + context = new AnnotationConfigApplicationContext(); + context.registerBean("theConnectionFactory", ConnectionFactory.class, () -> connectionFactory); + context.register(Config.class, TransactionalService.class); + context.refresh(); + + service = context.getBean(TransactionalService.class); + jdbc = createJdbcTemplate(createDataSource()); try { jdbc.execute("DROP TABLE legoset"); - } catch (DataAccessException e) {} + } catch (DataAccessException e) { + } jdbc.execute(getCreateTableStatement()); jdbc.execute("DELETE FROM legoset"); } + @After + public void tearDown() { + context.close(); + } + /** * Creates a {@link DataSource} to be used in this test. * @@ -97,6 +121,17 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend return "INSERT INTO legoset (id, name, manual) VALUES(:id, :name, :manual)"; } + /** + * Some Databases require special treatment to convince them to start a transaction. Some even start a transaction but + * store its id async so that it might show up a little late. + * + * @param client the client to use + * @return an empty {@link Mono} by default. + */ + protected Mono prepareForTransaction(DatabaseClient client) { + return Mono.empty(); + } + /** * Get a statement that returns the current transactionId. */ @@ -191,7 +226,8 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend .bind(0, 42055) // .bind(1, "SCHAUFELRADBAGGER") // .bindNull(2, Integer.class) // - .fetch().rowsUpdated().then(Mono.error(new IllegalStateException("failed"))); + .fetch().rowsUpdated() // + .then(Mono.error(new IllegalStateException("failed"))); }); integerFlux.as(StepVerifier::create) // @@ -202,7 +238,7 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend assertThat(count).isEqualTo(0); } - @Test // gh-2, gh-75 + @Test // gh-2, gh-75, gh-107 public void emitTransactionIds() { DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); @@ -210,22 +246,13 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend TransactionalOperator transactionalOperator = TransactionalOperator .create(new ConnectionFactoryTransactionManager(connectionFactory), new DefaultTransactionDefinition()); - // We have to execute a sql statement first. - // Otherwise some databases (MySql) don't have a transaction id. - Mono insert = databaseClient.execute().sql(getInsertIntoLegosetStatement()) // - .bind(0, 42055) // - .bind(1, "SCHAUFELRADBAGGER") // - .bindNull(2, Integer.class) // - .fetch().rowsUpdated(); - Flux txId = databaseClient.execute() // .sql(getCurrentTransactionIdStatement()) // .map((row, md) -> row.get(0)) // .all(); - // insert.thenMany fails because of a cancel signal. Probably a consequence of dematerialize - // in TransactionalOperator.execute. - Flux transactionIds = txId.concatWith(txId).as(transactionalOperator::transactional); + Flux transactionIds = prepareForTransaction(databaseClient).thenMany(txId.concatWith(txId)) // + .as(transactionalOperator::transactional); transactionIds.collectList().as(StepVerifier::create) // .consumeNextWith(actual -> { @@ -235,4 +262,114 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend }) // .verifyComplete(); } + + @Test // gh-107 + public void shouldRollbackTransactionUsingTransactionalOperator() { + + DatabaseClient databaseClient = DatabaseClient.create(connectionFactory); + + TransactionalOperator transactionalOperator = TransactionalOperator + .create(new ConnectionFactoryTransactionManager(connectionFactory), new DefaultTransactionDefinition()); + + Flux integerFlux = databaseClient.execute() // + .sql(getInsertIntoLegosetStatement()) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .thenMany(Mono.fromSupplier(() -> { + throw new IllegalStateException("failed"); + })); + + integerFlux.as(transactionalOperator::transactional) // + .as(StepVerifier::create) // + .expectError(IllegalStateException.class) // + .verify(); + + Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM legoset", Integer.class); + assertThat(count).isEqualTo(0); + } + + @Test //gh-107 + public void emitTransactionIdsUsingManagedTransactions() { + + service.emitTransactionIds(prepareForTransaction(service.getDatabaseClient()), getCurrentTransactionIdStatement()).collectList().as(StepVerifier::create) // + .consumeNextWith(actual -> { + + assertThat(actual).hasSize(2); + assertThat(actual.get(0)).isEqualTo(actual.get(1)); + }) // + .verifyComplete(); + } + + @Test // gh-107 + public void shouldRollbackTransactionUsingManagedTransactions() { + + service.shouldRollbackTransactionUsingTransactionalOperator(getInsertIntoLegosetStatement()) + .as(StepVerifier::create) // + .expectError(IllegalStateException.class) // + .verify(); + + Integer count = jdbc.queryForObject("SELECT COUNT(*) FROM legoset", Integer.class); + assertThat(count).isEqualTo(0); + } + + @Configuration + @EnableTransactionManagement + static class Config extends AbstractR2dbcConfiguration { + + @Autowired GenericApplicationContext context; + + @Override + public ConnectionFactory connectionFactory() { + return lookup(); + } + + ConnectionFactory lookup() { + return context.getBean("theConnectionFactory", ConnectionFactory.class); + } + + @Bean + ReactiveTransactionManager txMgr(ConnectionFactory connectionFactory) { + return new ConnectionFactoryTransactionManager(connectionFactory); + } + } + + static class TransactionalService { + + private DatabaseClient databaseClient; + + public TransactionalService(DatabaseClient databaseClient) { + this.databaseClient = databaseClient; + } + + @Transactional + public Flux emitTransactionIds(Mono prepareTransaction, String idStatement) { + + Flux txId = databaseClient.execute() // + .sql(idStatement) // + .map((row, md) -> row.get(0)) // + .all(); + + return prepareTransaction.thenMany(txId.concatWith(txId)); + } + + + @Transactional + public Flux shouldRollbackTransactionUsingTransactionalOperator(String insertStatement) { + + return databaseClient.execute().sql(insertStatement) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .thenMany(Mono.fromSupplier(() -> { + throw new IllegalStateException("failed"); + })); + } + + public DatabaseClient getDatabaseClient() { + return databaseClient; + } + } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java b/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java index 3efbc57..7519423 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java @@ -15,16 +15,16 @@ */ package org.springframework.data.r2dbc.function; -import io.r2dbc.spi.ConnectionFactory; - import javax.sql.DataSource; +import java.time.Duration; +import io.r2dbc.spi.ConnectionFactory; import org.junit.ClassRule; import org.junit.Ignore; import org.junit.Test; - import org.springframework.data.r2dbc.testing.ExternalDatabase; import org.springframework.data.r2dbc.testing.MySqlTestSupport; +import reactor.core.publisher.Mono; /** * Integration tests for {@link TransactionalDatabaseClient} against MySQL. @@ -51,6 +51,25 @@ public class MySqlTransactionalDatabaseClientIntegrationTests return MySqlTestSupport.CREATE_TABLE_LEGOSET; } + @Override + protected Mono prepareForTransaction(DatabaseClient client) { + + /* + * We have to execute a sql statement first. + * Otherwise MySql don't have a transaction id. + * And we need to delay emitting the result so that MySql has time to write the transaction id, which is done in + * batches every now and then. + * @see: https://dev.mysql.com/doc/refman/5.7/en/innodb-information-schema-internal-data.html + */ + return client.execute().sql(getInsertIntoLegosetStatement()) // + .bind(0, 42055) // + .bind(1, "SCHAUFELRADBAGGER") // + .bindNull(2, Integer.class) // + .fetch().rowsUpdated() // + .delayElement(Duration.ofMillis(50)) // + .then(); + } + @Override protected String getCurrentTransactionIdStatement() { return "SELECT tx.trx_id FROM information_schema.innodb_trx tx WHERE tx.trx_mysql_thread_id = connection_id()"; @@ -59,12 +78,6 @@ public class MySqlTransactionalDatabaseClientIntegrationTests @Override @Test @Ignore("MySQL creates transactions only on interaction with transactional tables. BEGIN does not create a txid") - public void shouldManageUserTransaction() {} - - @Override - @Test - @Ignore("Third element is cancelled, looks like a bug") - public void emitTransactionIds() { - super.emitTransactionIds(); + public void shouldManageUserTransaction() { } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java index ede09c9..b13fb55 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryTransactionManagerUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java index 002de90..b77a1a2 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/ConnectionFactoryUtilsUnitTests.java @@ -17,17 +17,20 @@ package org.springframework.data.r2dbc.function.connectionfactory; import static org.mockito.Mockito.*; +import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; -import reactor.test.StepVerifier; - +import org.assertj.core.api.Assertions; import org.junit.Test; - +import org.reactivestreams.Publisher; import org.springframework.transaction.NoTransactionException; +import reactor.core.publisher.Mono; +import reactor.test.StepVerifier; /** * Unit tests for {@link ConnectionFactoryUtils}. * * @author Mark Paluch + * @author Christoph Strobl */ public class ConnectionFactoryUtilsUnitTests { @@ -91,4 +94,21 @@ public class ConnectionFactoryUtilsUnitTests { .expectNext(factoryMock) // .verifyComplete(); } + + @Test // gh-107 + public void connectionFactoryRetunsConnectionWhenNoSyncronisationActive() { + + ConnectionFactory factoryMock = mock(ConnectionFactory.class); + Connection connection = mock(Connection.class); + Publisher p = Mono.just(connection); + doReturn(p).when(factoryMock).create(); + + ConnectionFactoryUtils.getConnection(factoryMock) // + .as(StepVerifier::create) // + .consumeNextWith(it -> { + Assertions.assertThat(it.getT1()).isEqualTo(connection); + Assertions.assertThat(it.getT2()).isEqualTo(factoryMock); + }) + .verifyComplete(); + } } diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java index 1ea4390..d1ad075 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/DelegatingConnectionFactoryUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, diff --git a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java index ecbdac0..6a578ee 100644 --- a/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java +++ b/src/test/java/org/springframework/data/r2dbc/function/connectionfactory/TransactionAwareConnectionFactoryProxyUnitTests.java @@ -5,7 +5,7 @@ * you may not use this file except in compliance with the License. * You may obtain a copy of the License at * - * http://www.apache.org/licenses/LICENSE-2.0 + * https://www.apache.org/licenses/LICENSE-2.0 * * Unless required by applicable law or agreed to in writing, software * distributed under the License is distributed on an "AS IS" BASIS, @@ -18,23 +18,22 @@ package org.springframework.data.r2dbc.function.connectionfactory; import static org.assertj.core.api.Assertions.*; import static org.mockito.Mockito.*; +import java.util.concurrent.atomic.AtomicReference; + import io.r2dbc.spi.Connection; import io.r2dbc.spi.ConnectionFactory; +import org.junit.Before; +import org.junit.Test; +import org.springframework.transaction.reactive.TransactionalOperator; import reactor.core.publisher.Mono; import reactor.test.StepVerifier; import reactor.util.function.Tuple2; -import java.util.concurrent.atomic.AtomicReference; - -import org.junit.Before; -import org.junit.Test; - -import org.springframework.transaction.reactive.TransactionalOperator; - /** * Unit tests for {@link TransactionAwareConnectionFactoryProxy}. * * @author Mark Paluch + * @author Christoph Strobl */ public class TransactionAwareConnectionFactoryProxyUnitTests { @@ -53,6 +52,96 @@ public class TransactionAwareConnectionFactoryProxyUnitTests { tm = new ConnectionFactoryTransactionManager(connectionFactoryMock); } + @Test // gh-107 + public void createShouldProxyConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .as(StepVerifier::create) // + .consumeNextWith(connection -> { + assertThat(connection).isInstanceOf(ConnectionProxy.class); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void unwrapShouldReturnTargetConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.unwrap()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void unwrapShouldReturnTargetConnectionEvenWhenClosed() { + + when(connectionMock1.close()).thenReturn(Mono.empty()); + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .flatMap(it -> Mono.from(it.close()).then(Mono.just(it))) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.unwrap()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void getTargetConnectionShouldReturnTargetConnection() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.getTargetConnection()).isEqualTo(connectionMock1); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void getTargetConnectionShouldThrowsErrorEvenWhenClosed() { + + when(connectionMock1.close()).thenReturn(Mono.empty()); + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .flatMap(it -> Mono.from(it.close()).then(Mono.just(it))) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThatExceptionOfType(IllegalStateException.class).isThrownBy(() -> proxy.getTargetConnection()); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void hashCodeShouldReturnProxyHash() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.hashCode()).isEqualTo(System.identityHashCode(proxy)); + }) + .verifyComplete(); + } + + @Test // gh-107 + public void equalsShouldCompareCorrectly() { + + new TransactionAwareConnectionFactoryProxy(connectionFactoryMock).create() // + .map(ConnectionProxy.class::cast) + .as(StepVerifier::create) // + .consumeNextWith(proxy -> { + assertThat(proxy.equals(proxy)).isTrue(); + assertThat(proxy.equals(connectionMock1)).isFalse(); + }) + .verifyComplete(); + } + @Test // gh-107 public void shouldEmitBoundConnection() { @@ -68,14 +157,13 @@ public class TransactionAwareConnectionFactoryProxyUnitTests { ConnectionFactoryUtils.getConnection(connectionFactoryMock).map(Tuple2::getT1) // .doOnNext(transactionalConnection::set).flatMap(it -> { - return proxyCf.create().doOnNext(connectionFromProxy -> { + return proxyCf.create().doOnNext(connectionFromProxy -> { - ConnectionProxy connectionProxy = (ConnectionProxy) connectionFromProxy; - assertThat(connectionProxy.getTargetConnection()).isSameAs(it); - assertThat(connectionProxy.unwrap()).isSameAs(it); - }); - - }).as(rxtx::transactional) // + ConnectionProxy connectionProxy = (ConnectionProxy) connectionFromProxy; + assertThat(connectionProxy.getTargetConnection()).isSameAs(it); + assertThat(connectionProxy.unwrap()).isSameAs(it); + }); + }).as(rxtx::transactional) // .flatMapMany(Connection::close) // .as(StepVerifier::create) // .verifyComplete(); diff --git a/src/test/resources/logback.xml b/src/test/resources/logback.xml index c9be4b4..32090ee 100644 --- a/src/test/resources/logback.xml +++ b/src/test/resources/logback.xml @@ -10,7 +10,7 @@ - +