#75 - Add support for MySQL using jasync-mysql.

We now support MySQL through the jasync-mysql driver that exposes its asynchronous functionality through a R2DBC wrapper implementation.

Jasync uses for now its own exceptions.

Original pull request: #84.

pom.xml

@@ -31,6 +31,8 @@
 		<degraph-check.version>0.1.4</degraph-check.version>
 		<hsqldb.version>2.4.1</hsqldb.version>
 		<postgresql.version>42.2.5</postgresql.version>
+		<mysql.version>5.1.47</mysql.version>
+		<jasync.version>0.9.38</jasync.version>
 		<mssql-jdbc.version>7.1.2.jre8-preview</mssql-jdbc.version>
 		<r2dbc-releasetrain.version>Arabba-M7</r2dbc-releasetrain.version>
 		<reactive-streams.version>1.0.1</reactive-streams.version>
@@ -173,6 +175,13 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>mysql</groupId>
+			<artifactId>mysql-connector-java</artifactId>
+			<version>${mysql.version}</version>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>com.microsoft.sqlserver</groupId>
 			<artifactId>mssql-jdbc</artifactId>
@@ -198,6 +207,13 @@
 			<scope>test</scope>
 		</dependency>
 
+		<dependency>
+			<groupId>com.github.jasync-sql</groupId>
+			<artifactId>jasync-r2dbc-mysql</artifactId>
+			<version>${jasync.version}</version>
+			<scope>test</scope>
+		</dependency>
+
 		<dependency>
 			<groupId>de.schauderhaft.degraph</groupId>
 			<artifactId>degraph-check</artifactId>
@@ -295,7 +311,8 @@
 						<querydslVersion>${querydsl}</querydslVersion>
 						<springVersion>${spring}</springVersion>
 						<r2dbcVersion>${r2dbc-spi.version}</r2dbcVersion>
-						<reactiveStreamsVersion>${reactive-streams.version}</reactiveStreamsVersion>
+						<reactiveStreamsVersion>${reactive-streams.version}
+						</reactiveStreamsVersion>
 						<releasetrainVersion>${releasetrain}</releasetrainVersion>
 						<allow-uri-read>true</allow-uri-read>
 						<toclevels>3</toclevels>
@@ -416,6 +433,10 @@
 			<id>spring-libs-snapshot</id>
 			<url>https://repo.spring.io/libs-snapshot</url>
 		</repository>
+		<repository>
+			<id>jcenter</id>
+			<url>https://jcenter.bintray.com/</url>
+		</repository>
 	</repositories>
 
 	<pluginRepositories>

src/main/java/org/springframework/data/r2dbc/dialect/AnonymousBindMarkers.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/main/java/org/springframework/data/r2dbc/dialect/Database.java

@@ -53,6 +53,18 @@ public enum Database {
 		public Dialect defaultDialect() {
 			return H2Dialect.INSTANCE;
 		}
+	},
+
+	MYSQL {
+		@Override
+		public String driverName() {
+			return "MySQL";
+		}
+
+		@Override
+		public Dialect defaultDialect() {
+			return MySqlDialect.INSTANCE;
+		}
 	};
 
 	/**

src/main/java/org/springframework/data/r2dbc/dialect/IndexedBindMarker.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/main/java/org/springframework/data/r2dbc/dialect/MySqlDialect.java

@@ -0,0 +1,109 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.dialect;
+
+import java.net.InetAddress;
+import java.net.URI;
+import java.net.URL;
+import java.util.Arrays;
+import java.util.Collection;
+import java.util.HashSet;
+import java.util.Set;
+import java.util.UUID;
+
+/**
+ * An SQL dialect for MySQL.
+ *
+ * @author Mark Paluch
+ */
+public class MySqlDialect implements Dialect {
+
+	private static final Set<Class<?>> SIMPLE_TYPES = new HashSet<>(
+			Arrays.asList(UUID.class, URL.class, URI.class, InetAddress.class));
+
+	/**
+	 * Singleton instance.
+	 */
+	public static final MySqlDialect INSTANCE = new MySqlDialect();
+
+	private static final BindMarkersFactory ANONYMOUS = BindMarkersFactory.anonymous("?");
+
+	private static final LimitClause LIMIT_CLAUSE = new LimitClause() {
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long, long)
+		 */
+		@Override
+		public String getClause(long limit, long offset) {
+			return String.format("LIMIT %d,%d", limit, offset);
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClause(long)
+		 */
+		@Override
+		public String getClause(long limit) {
+			return "LIMIT " + limit;
+		}
+
+		/*
+		 * (non-Javadoc)
+		 * @see org.springframework.data.r2dbc.dialect.LimitClause#getClausePosition()
+		 */
+		@Override
+		public Position getClausePosition() {
+			return Position.END;
+		}
+	};
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getBindMarkersFactory()
+	 */
+	@Override
+	public BindMarkersFactory getBindMarkersFactory() {
+		return ANONYMOUS;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getSimpleTypesKeys()
+	 */
+	@Override
+	public Collection<? extends Class<?>> getSimpleTypes() {
+		return SIMPLE_TYPES;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#limit()
+	 */
+	@Override
+	public LimitClause limit() {
+		return LIMIT_CLAUSE;
+	}
+
+	/*
+	 * (non-Javadoc)
+	 * @see org.springframework.data.r2dbc.dialect.Dialect#getArraySupport()
+	 */
+	@Override
+	public ArrayColumns getArraySupport() {
+		return ArrayColumns.Unsupported.INSTANCE;
+	}
+}

src/test/java/org/springframework/data/r2dbc/dialect/AnonymousBindMarkersUnitTests.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/AbstractTransactionalDatabaseClientIntegrationTests.java

@@ -32,6 +32,7 @@ import javax.sql.DataSource;
 
 import org.junit.Before;
 import org.junit.Test;
+
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.r2dbc.testing.R2dbcIntegrationTestSupport;
 import org.springframework.jdbc.core.JdbcTemplate;
@@ -201,15 +202,21 @@ public abstract class AbstractTransactionalDatabaseClientIntegrationTests extend
 		assertThat(count).isEqualTo(0);
 	}
 
-	@Test // gh-2
+	@Test // gh-2, gh-75
 	public void emitTransactionIds() {
 
 		TransactionalDatabaseClient databaseClient = TransactionalDatabaseClient.create(connectionFactory);
 
 		Flux<Object> transactionIds = databaseClient.inTransaction(db -> {
 
+			Mono<Integer> insert = db.execute().sql(getInsertIntoLegosetStatement()) //
+					.bind(0, 42055) //
+					.bind(1, "SCHAUFELRADBAGGER") //
+					.bindNull(2, Integer.class) //
+					.fetch().rowsUpdated();
+
 			Flux<Object> txId = db.execute().sql(getCurrentTransactionIdStatement()).map((r, md) -> r.get(0)).all();
-			return txId.concatWith(txId);
+			return insert.thenMany(txId.concatWith(txId));
 		});
 
 		transactionIds.collectList().as(StepVerifier::create) //

src/test/java/org/springframework/data/r2dbc/function/MySqlDatabaseClientIntegrationTests.java

@@ -0,0 +1,57 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import javax.sql.DataSource;
+
+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;
+
+/**
+ * Integration tests for {@link DatabaseClient} against MySQL.
+ *
+ * @author Mark Paluch
+ */
+public class MySqlDatabaseClientIntegrationTests extends AbstractDatabaseClientIntegrationTests {
+
+	@ClassRule public static final ExternalDatabase database = MySqlTestSupport.database();
+
+	@Override
+	protected DataSource createDataSource() {
+		return MySqlTestSupport.createDataSource(database);
+	}
+
+	@Override
+	protected ConnectionFactory createConnectionFactory() {
+		return MySqlTestSupport.createConnectionFactory(database);
+	}
+
+	@Override
+	protected String getCreateTableStatement() {
+		return MySqlTestSupport.CREATE_TABLE_LEGOSET;
+	}
+
+	@Override
+	@Ignore("Jasync currently uses its own exceptions, see jasync-sql/jasync-sql#106")
+	@Test
+	public void shouldTranslateDuplicateKeyException() {}
+}

src/test/java/org/springframework/data/r2dbc/function/MySqlTransactionalDatabaseClientIntegrationTests.java

@@ -0,0 +1,63 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.function;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import javax.sql.DataSource;
+
+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;
+
+/**
+ * Integration tests for {@link TransactionalDatabaseClient} against MySQL.
+ *
+ * @author Mark Paluch
+ */
+public class MySqlTransactionalDatabaseClientIntegrationTests
+		extends AbstractTransactionalDatabaseClientIntegrationTests {
+
+	@ClassRule public static final ExternalDatabase database = MySqlTestSupport.database();
+
+	@Override
+	protected DataSource createDataSource() {
+		return MySqlTestSupport.createDataSource(database);
+	}
+
+	@Override
+	protected ConnectionFactory createConnectionFactory() {
+		return MySqlTestSupport.createConnectionFactory(database);
+	}
+
+	@Override
+	protected String getCreateTableStatement() {
+		return MySqlTestSupport.CREATE_TABLE_LEGOSET;
+	}
+
+	@Override
+	protected String getCurrentTransactionIdStatement() {
+		return "SELECT tx.trx_id FROM information_schema.innodb_trx tx WHERE tx.trx_mysql_thread_id = connection_id()";
+	}
+
+	@Override
+	@Test
+	@Ignore("MySQL creates transactions only on interaction with transactional tables. BEGIN does not create a txid")
+	public void shouldManageUserTransaction() {}
+}

src/test/java/org/springframework/data/r2dbc/repository/MySqlR2dbcRepositoryIntegrationTests.java

@@ -0,0 +1,99 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.repository;
+
+import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+import javax.sql.DataSource;
+
+import org.junit.ClassRule;
+import org.junit.runner.RunWith;
+
+import org.springframework.context.annotation.ComponentScan.Filter;
+import org.springframework.context.annotation.Configuration;
+import org.springframework.context.annotation.FilterType;
+import org.springframework.data.r2dbc.config.AbstractR2dbcConfiguration;
+import org.springframework.data.r2dbc.repository.config.EnableR2dbcRepositories;
+import org.springframework.data.r2dbc.repository.query.Query;
+import org.springframework.data.r2dbc.repository.support.R2dbcRepositoryFactory;
+import org.springframework.data.r2dbc.testing.ExternalDatabase;
+import org.springframework.data.r2dbc.testing.MySqlTestSupport;
+import org.springframework.test.context.ContextConfiguration;
+import org.springframework.test.context.junit4.SpringRunner;
+
+/**
+ * Integration tests for {@link LegoSetRepository} using {@link R2dbcRepositoryFactory} against MySQL.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(SpringRunner.class)
+@ContextConfiguration
+public class MySqlR2dbcRepositoryIntegrationTests extends AbstractR2dbcRepositoryIntegrationTests {
+
+	@ClassRule public static final ExternalDatabase database = MySqlTestSupport.database();
+
+	@Configuration
+	@EnableR2dbcRepositories(considerNestedRepositories = true,
+			includeFilters = @Filter(classes = MySqlLegoSetRepository.class, type = FilterType.ASSIGNABLE_TYPE))
+	static class IntegrationTestConfiguration extends AbstractR2dbcConfiguration {
+
+		@Override
+		public ConnectionFactory connectionFactory() {
+			return MySqlTestSupport.createConnectionFactory(database);
+		}
+	}
+
+	@Override
+	protected DataSource createDataSource() {
+		return MySqlTestSupport.createDataSource(database);
+	}
+
+	@Override
+	protected ConnectionFactory createConnectionFactory() {
+		return MySqlTestSupport.createConnectionFactory(database);
+	}
+
+	@Override
+	protected String getCreateTableStatement() {
+		return MySqlTestSupport.CREATE_TABLE_LEGOSET_WITH_ID_GENERATION;
+	}
+
+	@Override
+	protected Class<? extends LegoSetRepository> getRepositoryInterfaceType() {
+		return MySqlLegoSetRepository.class;
+	}
+
+	interface MySqlLegoSetRepository extends LegoSetRepository {
+
+		@Override
+		@Query("SELECT * FROM legoset WHERE name like ?")
+		Flux<LegoSet> findByNameContains(String name);
+
+		@Override
+		@Query("SELECT * FROM legoset")
+		Flux<Named> findAsProjection();
+
+		@Override
+		@Query("SELECT * FROM legoset WHERE manual = :manual")
+		Mono<LegoSet> findByManual(int manual);
+
+		@Override
+		@Query("SELECT id FROM legoset")
+		Flux<Integer> findAllIds();
+	}
+}

src/test/java/org/springframework/data/r2dbc/testing/MySqlTestSupport.java

@@ -0,0 +1,151 @@
+/*
+ * 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.
+ * You may obtain a copy of the License at
+ *
+ *      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,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+package org.springframework.data.r2dbc.testing;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import java.util.function.Supplier;
+import java.util.stream.Stream;
+
+import javax.sql.DataSource;
+
+import org.springframework.data.r2dbc.testing.ExternalDatabase.ProvidedDatabase;
+
+import org.testcontainers.containers.MySQLContainer;
+
+import com.github.jasync.r2dbc.mysql.JasyncConnectionFactory;
+import com.github.jasync.sql.db.Configuration;
+import com.github.jasync.sql.db.mysql.pool.MySQLConnectionFactory;
+import com.mysql.jdbc.jdbc2.optional.MysqlDataSource;
+
+/**
+ * Utility class for testing against MySQL.
+ *
+ * @author Mark Paluch
+ */
+public class MySqlTestSupport {
+
+	private static ExternalDatabase testContainerDatabase;
+
+	public static String CREATE_TABLE_LEGOSET = "CREATE TABLE legoset (\n" //
+			+ "    id          integer PRIMARY KEY,\n" //
+			+ "    name        varchar(255) NOT NULL,\n" //
+			+ "    manual      integer NULL\n" //
+			+ ") ENGINE=InnoDB;";
+
+	public static String CREATE_TABLE_LEGOSET_WITH_ID_GENERATION = "CREATE TABLE legoset (\n" //
+			+ "    id          integer AUTO_INCREMENT PRIMARY KEY,\n" //
+			+ "    name        varchar(255) NOT NULL,\n" //
+			+ "    manual      integer NULL\n" //
+			+ ") ENGINE=InnoDB;";
+
+	/**
+	 * Returns a database either hosted locally at {@code postgres:@localhost:5432/postgres} or running inside Docker.
+	 *
+	 * @return information about the database. Guaranteed to be not {@literal null}.
+	 */
+	public static ExternalDatabase database() {
+
+		if (Boolean.getBoolean("spring.data.r2dbc.test.preferLocalDatabase")) {
+
+			return getFirstWorkingDatabase( //
+					MySqlTestSupport::local, //
+					MySqlTestSupport::testContainer //
+			);
+		} else {
+
+			return getFirstWorkingDatabase( //
+					MySqlTestSupport::testContainer, //
+					MySqlTestSupport::local //
+			);
+		}
+	}
+
+	@SafeVarargs
+	private static ExternalDatabase getFirstWorkingDatabase(Supplier<ExternalDatabase>... suppliers) {
+
+		return Stream.of(suppliers).map(Supplier::get) //
+				.filter(ExternalDatabase::checkValidity) //
+				.findFirst() //
+				.orElse(ExternalDatabase.unavailable());
+	}
+
+	/**
+	 * Returns a locally provided database at {@code postgres:@localhost:5432/postgres}.
+	 */
+	private static ExternalDatabase local() {
+
+		return ProvidedDatabase.builder() //
+				.hostname("localhost") //
+				.port(3306) //
+				.database("mysql") //
+				.username("root") //
+				.password("my-secret-pw").build();
+	}
+
+	/**
+	 * Returns a database provided via Testcontainers.
+	 */
+	private static ExternalDatabase testContainer() {
+
+		if (testContainerDatabase == null) {
+
+			try {
+				MySQLContainer mySQLContainer = new MySQLContainer("mysql:5.6.43");
+				mySQLContainer.start();
+
+				testContainerDatabase = ProvidedDatabase.builder() //
+						.hostname("localhost") //
+						.port(mySQLContainer.getFirstMappedPort()) //
+						.database(mySQLContainer.getDatabaseName()) //
+						.username("root") //
+						.password(mySQLContainer.getPassword()).build();
+			} catch (IllegalStateException ise) {
+				// docker not available.
+				testContainerDatabase = ExternalDatabase.unavailable();
+			}
+		}
+
+		return testContainerDatabase;
+	}
+
+	/**
+	 * Creates a new {@link ConnectionFactory} configured from the {@link ExternalDatabase}..
+	 */
+	public static ConnectionFactory createConnectionFactory(ExternalDatabase database) {
+
+		MySQLConnectionFactory jasync = new MySQLConnectionFactory(new Configuration(database.getUsername(),
+				database.getHostname(), database.getPort(), database.getPassword(), database.getDatabase()));
+		return new JasyncConnectionFactory(jasync);
+	}
+
+	/**
+	 * Creates a new {@link DataSource} configured from the {@link ExternalDatabase}.
+	 */
+	public static DataSource createDataSource(ExternalDatabase database) {
+
+		MysqlDataSource dataSource = new MysqlDataSource();
+
+		dataSource.setUser(database.getUsername());
+		dataSource.setPassword(database.getPassword());
+		dataSource.setDatabaseName(database.getDatabase());
+		dataSource.setServerName(database.getHostname());
+		dataSource.setPortNumber(database.getPort());
+
+		return dataSource;
+	}
+
+}