#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.

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 String driverName() {
 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 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 void shouldRollbackTransaction() {
 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/AbstractR2dbcRepositoryIntegrationTests.java

@@ -188,16 +188,16 @@ public void shouldInsertItemsTransactional() {
 Flux<Map<String, Object>> transactional = client.inTransaction(db -> {
 
 return transactionalRepository.save(legoSet1) //
-.map(it -> jdbc.queryForMap("SELECT count(*) FROM legoset"));
+.map(it -> jdbc.queryForMap("SELECT count(*) as count FROM legoset"));
 });
 
 Mono<Map<String, Object>> nonTransactional = transactionalRepository.save(legoSet2) //
-.map(it -> jdbc.queryForMap("SELECT count(*) FROM legoset"));
+.map(it -> jdbc.queryForMap("SELECT count(*) as count FROM legoset"));
 
 transactional.as(StepVerifier::create).expectNext(Collections.singletonMap("count", 0L)).verifyComplete();
 nonTransactional.as(StepVerifier::create).expectNext(Collections.singletonMap("count", 2L)).verifyComplete();
 
-Map<String, Object> count = jdbc.queryForMap("SELECT count(*) FROM legoset");
+Map<String, Object> count = jdbc.queryForMap("SELECT count(*) as count FROM legoset");
 assertThat(count).containsEntry("count", 2L);
 }