#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

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<Void> 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<Integer> insert = databaseClient.execute().sql(getInsertIntoLegosetStatement()) //
-				.bind(0, 42055) //
-				.bind(1, "SCHAUFELRADBAGGER") //
-				.bindNull(2, Integer.class) //
-				.fetch().rowsUpdated();
-
 		Flux<Object> 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<Object> transactionIds = txId.concatWith(txId).as(transactionalOperator::transactional);
+		Flux<Object> 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<Integer> 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<Object> emitTransactionIds(Mono<Void> prepareTransaction, String idStatement) {
+
+			Flux<Object> txId = databaseClient.execute() //
+					.sql(idStatement) //
+					.map((row, md) -> row.get(0)) //
+					.all();
+
+			return prepareTransaction.thenMany(txId.concatWith(txId));
+		}
+
+
+		@Transactional
+		public Flux<Integer> 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;
+		}
+	}
 }

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<Void> 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() {
 	}
 }

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,

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<? extends Connection> 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();
+	}
 }

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,

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();