Reactive Relational Database Connectivity MySQL Implementation
This project contains the MySQL implementation of the R2DBC SPI.
This implementation is not intended to be used directly, but rather to be
used as the backing implementation for a humane client library to
delegate to. See R2DBC Homepage.
This driver provides the following features:
Unix domain socket.
Execution of simple or batch statements without bindings.
Execution of prepared statements with bindings.
Reactive LOB types (e.g. BLOB, CLOB)
All charsets from MySQL, like utf8mb4_0900_ai_ci, latin1_general_ci, utf32_unicode_520_ci, etc.
All authentication types for MySQL, like caching_sha2_password, mysql_native_password, etc.
General exceptions with error code and standard SQL state mappings.
Secure connection with verification (SSL/TLS), auto-select TLS version for community and enterprise editions.
SSL tunnel for proxy protocol of MySQL.
Transactions with savepoint.
Native ping command that can be verifying when argument is ValidationDepth.REMOTE
Extensible, e.g. extend built-in Codec(s).
Version compatibility / Integration tests states
In fact, it supports lower versions, in the theory, such as 4.1, 4.0, etc.
However, Docker-certified images do not have these versions lower than 5.5.0, so tests are not integrated on these versions.
If you'd rather like the latest snapshots of the upcoming major version, use SonaType Maven snapshot repository and declare the appropriate dependency version.
dependencies {
// Maybe should to use `compile` instead of `implementation` on the lower version of Gradle.
implementation("dev.miku:r2dbc-mysql:0.8.2.RELEASE")
}
Getting Started
Here is a quick teaser of how to use R2DBC MySQL in Java:
URL Connection Factory Discovery
// Notice: the query string must be URL encodedConnectionFactoryconnectionFactory= ConnectionFactories.get(
"r2dbcs:mysql://root:database-password-in-here@127.0.0.1:3306/r2dbc?" +
"zeroDate=use_round&" +
"sslMode=verify_identity&" +
"useServerPrepareStatement=true&" +
"tlsVersion=TLSv1.3%2CTLSv1.2%2CTLSv1.1&" +
"sslCa=%2Fpath%2Fto%2Fmysql%2Fca.pem&" +
"sslKey=%2Fpath%2Fto%2Fmysql%2Fclient-key.pem&" +
"sslCert=%2Fpath%2Fto%2Fmysql%2Fclient-cert.pem&" +
"sslKeyPassword=key-pem-password-in-here"
)
// Creating a Mono using Project Reactor
Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
It is just example, see also Programmatic Connection Factory Discovery for more options.
ConnectionFactoryOptionsoptions= ConnectionFactoryOptions.builder()
.option(DRIVER, "mysql")
.option(HOST, "127.0.0.1")
.option(USER, "root")
.option(PORT, 3306) // optional, default 3306
.option(PASSWORD, "database-password-in-here") // optional, default null, null means has no password
.option(DATABASE, "r2dbc") // optional, default null, null means not specifying the database
.option(CONNECT_TIMEOUT, Duration.ofSeconds(3)) // optional, default null, null means no timeout
.option(SSL, true) // optional, default sslMode is "preferred", it will be ignore if sslMode is set
.option(Option.valueOf("sslMode"), "verify_identity") // optional, default "preferred"
.option(Option.valueOf("sslCa"), "/path/to/mysql/ca.pem") // required when sslMode is verify_ca or verify_identity, default null, null means has no server CA cert
.option(Option.valueOf("sslCert"), "/path/to/mysql/client-cert.pem") // optional, default null, null means has no client cert
.option(Option.valueOf("sslKey"), "/path/to/mysql/client-key.pem") // optional, default null, null means has no client key
.option(Option.valueOf("sslKeyPassword"), "key-pem-password-in-here") // optional, default null, null means has no password for client key (i.e. "sslKey")
.option(Option.valueOf("tlsVersion"), "TLSv1.3,TLSv1.2,TLSv1.1") // optional, default is auto-selected by the server
.option(Option.valueOf("sslHostnameVerifier"), "com.example.demo.MyVerifier") // optional, default is null, null means use standard verifier
.option(Option.valueOf("sslContextBuilderCustomizer"), "com.example.demo.MyCustomizer") // optional, default is no-op customizer
.option(Option.valueOf("zeroDate"), "use_null") // optional, default "use_null"
.option(Option.valueOf("useServerPrepareStatement"), true) // optional, default false
.option(Option.valueOf("tcpKeepAlive"), true) // optional, default false
.option(Option.valueOf("tcpNoDelay"), true) // optional, default false
.option(Option.valueOf("autodetectExtensions"), false) // optional, default false
.build();
ConnectionFactoryconnectionFactory= ConnectionFactories.get(options);
// Creating a Mono using Project Reactor
Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
MySqlConnectionConfigurationconfiguration= MySqlConnectionConfiguration.builder()
.host("127.0.0.1")
.user("root")
.port(3306) // optional, default 3306
.password("database-password-in-here") // optional, default null, null means has no password
.database("r2dbc") // optional, default null, null means not specifying the database
.serverZoneId(ZoneId.of("Continent/City")) // optional, default null, null means query server time zone when connection init
.connectTimeout(Duration.ofSeconds(3)) // optional, default null, null means no timeout
.sslMode(SslMode.VERIFY_IDENTITY) // optional, default SslMode.PREFERRED
.sslCa("/path/to/mysql/ca.pem") // required when sslMode is VERIFY_CA or VERIFY_IDENTITY, default null, null means has no server CA cert
.sslCert("/path/to/mysql/client-cert.pem") // optional, default has no client SSL certificate
.sslKey("/path/to/mysql/client-key.pem") // optional, default has no client SSL key
.sslKeyPassword("key-pem-password-in-here") // optional, default has no client SSL key password
.tlsVersion(TlsVersions.TLS1_3, TlsVersions.TLS1_2, TlsVersions.TLS1_1) // optional, default is auto-selected by the server
.sslHostnameVerifier(MyVerifier.INSTANCE) // optional, default is null, null means use standard verifier
.sslContextBuilderCustomizer(MyCustomizer.INSTANCE) // optional, default is no-op customizer
.zeroDateOption(ZeroDateOption.USE_NULL) // optional, default ZeroDateOption.USE_NULL
.useServerPrepareStatement() // Use server-preparing statements, default use client-preparing statements
.tcpKeepAlive(true) // optional, controls TCP Keep Alive, default is false
.tcpNoDelay(true) // optional, controls TCP No Delay, default is false
.autodetectExtensions(false) // optional, controls extension auto-detect, default is true
.extendWith(MyExtension.INSTANCE) // optional, manual extend an extension into extensions, default using auto-detect
.build();
ConnectionFactoryconnectionFactory= MySqlConnectionFactory.from(configuration);
// Creating a Mono using Project Reactor
Mono<Connection> connectionMono = Mono.from(connectionFactory.create());
This driver needs to be discovered by name in R2DBC
host
A hostname or IP
Required when unixSocket does not exists
The host of MySQL database server
unixSocket
An absolute or relative path
Required when host does not exists
The .sock file of Unix Domain Socket
port
A positive integer less than 65536
Optional, default 3306
The port of MySQL database server
user
A valid MySQL username and not be empty
Required
Who wants to connect to the MySQL database
password
Any printable string
Optional, default no password
The password of the MySQL database user
database
A valid MySQL database name
Optional, default does not initialize database
Database used by the MySQL connection
connectTimeout
A Duration which must be positive duration
Optional, default has no timeout
TCP connect timeout
serverZoneId
An id of ZoneId
Optional, default query time zone when connection init
Server time zone id
tcpKeepAlive
true or false
Optional, default disabled
Controls TCP KeepAlive
tcpNoDelay
true or false
Optional, default disabled
Controls TCP NoDelay
sslMode
A value of SslMode
Optional, default PREFERRED when using hosting connection, DISABLED when using Unix Domain Socket
SSL mode, see following notice
sslCa
A path of local file which type is PEM
Required when sslMode is VERIFY_CA or VERIFY_IDENTITY
The CA cert of MySQL database server
sslCert
A path of local file which type is PEM
Required when sslKey exists
The SSL cert of client
sslKey
A path of local file which type is PEM
Required when sslCert exists
The SSL key of client
sslKeyPassword
Any valid password for PEM file
Optional, default sslKey has no password
The password for client SSL key (i.e. sslKey)
tlsVersion
Any value list of TlsVersions
Optional, default is auto-selected by the server
The TLS version for SSL, see following notice
sslHostnameVerifier
A HostnameVerifier
Optional, default use RFC standard
Used only if SslMode is VERIFY_CA or higher
sslContextBuilderCustomizer
A Function<SslContextBuilder, SslContextBuilder>
Optional, default is NO-OP function
Used only if SslMode is not DISABLED
zeroDateOption
Any value of ZeroDateOption
Optional, default USE_NULL
The option indicates "zero date" handling, see following notice
autodetectExtensions
true or false
Optional, default is true
Controls auto-detect Extensions
useServerPrepareStatement
true, false or Predicate<String>
Optional, default is false
See following notice
SslMode Considers security level and verification for SSL, make sure the database server supports SSL before you want change SSL mode to REQUIRED or higher. The Unix Domain Socket only offers "DISABLED" available
DISABLED I don't care about security and don't want to pay the overhead for encryption
PREFERRED I don't care about encryption but will pay the overhead of encryption if the server supports it. Unavailable on Unix Domain Socket
REQUIRED I want my data to be encrypted, and I accept the overhead. I trust that the network will make sure I always connect to the server I want. Unavailable on Unix Domain Socket
VERIFY_CA I want my data encrypted, and I accept the overhead. I want to be sure I connect to a server that I trust. Unavailable on Unix Domain Socket
VERIFY_IDENTITY (the highest level, most like web browser): I want my data encrypted, and I accept the overhead. I want to be sure I connect to a server I trust, and that it's the one I specify. Unavailable on Unix Domain Socket
TUNNEL Use SSL tunnel to connect to MySQL, it may be useful for some RDS that's using SSL proxy Unavailable on Unix Domain Socket
TlsVersions Considers TLS version names for SSL, can be multi-values in the configuration, make sure the database server supports selected TLS versions. Usually sorted from higher to lower. Unavailable on Unix Domain Socket
TLS1 (i.e. "TLSv1") Under generic circumstances, MySQL database supports it if database supports SSL
TLS1_1 (i.e. "TLSv1.1") Under generic circumstances, MySQL database supports it if database supports SSL
TLS1_2 (i.e. "TLSv1.2") Supported only in Community Edition 8.0.4 or higher, and Enterprise Edition 5.6.0 or higher
TLS1_3 (i.e. "TLSv1.3") Supported only available as of MySQL 8.0.16 or higher, and requires compiling MySQL using OpenSSL 1.1.1 or higher
ZeroDateOption Considers special handling when MySQL database server returning "zero date" (i.e. 0000-00-00 00:00:00)
EXCEPTION Just throw an exception when MySQL database server return "zero date"
USE_NULL Use null when MySQL database server return "zero date"
USE_ROUNDNOT RECOMMENDED, only for compatibility. Use "round" date (i.e. 0001-01-01 00:00:00) when MySQL database server return "zero date"
Prepare Statement: Considers based on server-preparing or client-preparing, some database server maybe not support server-preparing binary-query, such as Vitess
useClientPrepareStatement() default preparing mode, use client-preparing text-query for parametrized statements
useServerPrepareStatement() use server-preparing binary-query for parametrized statements
useServerPrepareStatement(Predicate<String>) use server-preparing binary-query for parametrized statements, and enforce server-preparing usage for simple query (not parametrized statements). The usage is judged by Predicate, it's parameter is the simple SQL statement, enforce server-preparing if return true
extendWith Manual extend Extension, only available in programmatic configuration
It is NOT RECOMMENDED, enable the autodetectExtensions is the best way for extensions
The Extensions will not remove duplicates, make sure it would be not extended twice or more
The auto-detected Extensions will not affect manual extends and will not remove duplicates
Should use enum in Programmatic configuration that not like discovery configurations, except TlsVersions (All elements of TlsVersions will be always String which is case-sensitive).
connection.createStatement("INSERT INTO `person` (`first_name`, `last_name`) VALUES ('who', 'how')")
.execute(); // return a Publisher include one Result
Parametrized statement
connection.createStatement("INSERT INTO `person` (`birth`, `nickname`, `show_name`) VALUES (?, ?name, ?name)")
.bind(0, LocalDateTime.of(2019, 6, 25, 12, 12, 12))
.bind("name", "Some one") // Not one-to-one binding, call twice of native index-bindings, or call once of name-bindings.
.add()
.bind(0, LocalDateTime.of(2009, 6, 25, 12, 12, 12))
.bind(1, "My Nickname")
.bind(2, "Naming show")
.returnGeneratedValues("generated_id")
.execute(); // return a Publisher include two Results.
All parameters must be bound before execute, even parameter is null (use bindNull to bind null).
It will be using client-preparing by default, see useServerPrepareStatement in configuration.
In one-to-one binding, because native MySQL prepared statements use index-based parameters, index-bindings will have better performance than name-bindings.
Batch statement
connection.createBatch()
.add("INSERT INTO `person` (`first_name`, `last_name`) VALUES ('who', 'how')")
.add("UPDATE `earth` SET `count` = `count` + 1 WHERE `id` = 'human'")
.execute(); // return a Publisher include two Results.
The last ; will be removed if and only if last statement contains ';', and statement has only whitespace follow the last ;.
This is an extension of a highly customized driver behavior of encoding parameter or decoding field data.
Example for an extending Codec of JSON based-on Jackson.
First, implement a Codec, ParametrizedCodec, MassiveCodec or MassiveParametrizedCodec.
Codec the normal codec
Data type is Class<?>
Data buffer size is less than or equal to Integer.MAX_VALUE
ParametrizedCodec
Data type is Class<?> or ParametrizedType
Data buffer size is less than or equal to Integer.MAX_VALUE
MassiveCodec
Data type is Class<?>
Data buffer size is less than or equal to UnsignedInteger.MAX_VALUE (Java does not have unsigned integers, it just only represents the range)
MassiveParametrizedCodec
Data type is Class<?> or ParametrizedType
Data buffer size is less than or equal to UnsignedInteger.MAX_VALUE
Actually, JSON can store large json data, and its byte size can be UnsignedInteger.MAX_VALUE. However, this is just an example.
publicfinalclassJacksonCodecimplementsCodec<Object> {
/**
* JUST for example, should configure it in real applications.
*/privatestaticfinalObjectMapperMAPPER=newObjectMapper();
privatefinal ByteBufAllocator allocator;
/**
* Used for encoding/decoding mode, see also registrar in second step.
*/privatefinalboolean encoding;
publicJacksonCodec(ByteBufAllocator allocator, boolean encoding) {
this.allocator = allocator;
this.encoding = encoding;
}
@Overridepublic Object decode(ByteBuf value, FieldInformation info, Class<?> target, boolean binary, CodecContext context) {
// If you ensure server is using UTF-8, you can just use InputStreamtry (Readerr=newInputStreamReader(newByteBufInputStream(value), CharCollation.fromId(info.getCollationId(), context.getServerVersion()).getCharset())) {
return MAPPER.readValue(r, target);
} catch (IOException e) {
thrownewRuntimeException(e);
}
}
@Overridepublic Parameter encode(Object value, CodecContext context) {
returnnewJacksonParameter(allocator, value, context);
}
@OverridepublicbooleancanDecode(FieldInformation info, Class<?> target) {
return !encoding && info.getType() == DataTypes.JSON && info.getCollationId() != CharCollation.BINARY_ID;
}
@OverridepublicbooleancanEncode(Object value) {
return encoding;
}
privatestaticfinalclassJacksonParameterimplementsParameter {
privatefinal ByteBufAllocator allocator;
privatefinal Object value;
privatefinal CodecContext context;
privateJacksonParameter(ByteBufAllocator allocator, Object value, CodecContext context) {
this.allocator = allocator;
this.value = value;
this.context = context;
}
@Overridepublic Mono<ByteBuf> publishBinary() {
// JSON in binary protocol should be a var-integer sized encoded string.// That means we should write a var-integer as a size of following content// bytes firstly, then write the encoded string as content.//// Binary protocol may be different for each type of encoding, so if do not// use binary protocol, just return a Mono.error() instead.return Mono.fromSupplier(() -> {
Charsetcharset= context.getClientCollation().getCharset();
ByteBufcontent= allocator.buffer();
// Encode and calculate content bytes first, we should know bytes size.try (Writerw=newOutputStreamWriter(newByteBufOutputStream(content), charset)) {
MAPPER.writeValue(w, value);
} catch (IOException e) {
content.release();
thrownewCustomRuntimeException(e);
} catch (Throwable e) {
content.release();
throw e;
}
ByteBufbuf=null;
try {
buf = allocator.buffer();
// VarIntUtils is an unstable, internal utility.
VarIntUtils.writeVarInt(buf, content.readableBytes());
return buf.writeBytes(content);
} catch (Throwable e) {
if (buf != null) {
buf.release();
}
throw e;
} finally {
content.release();
}
});
}
@Overridepublic Mono<Void> publishText(ParameterWriter writer) {
return Mono.fromRunnable(() -> {
try {
MAPPER.writeValue(writer, value);
} catch (IOException e) {
thrownewCustomRuntimeException(e);
}
});
}
@OverridepublicshortgetType() {
return DataTypes.VARCHAR;
}
}
}
Second, implement a CodecRegistrar.
// It is just an example of package name and does not represent any company, individual or organization.package org.example.demo.json;
// Some imports...publicfinalclassJacksonCodecRegistrarimplementsCodecRegistrar {
@Overridepublicvoidregister(ByteBufAllocator allocator, CodecRegistry registry) {
// Decoding JSON by highest priority, encoding anything by lowest priority.
registry.addFirst(newJacksonCodec(allocator, false))
.addLast(newJacksonCodec(allocator, true));
}
}
Finally, create a file in META-INF/services, which file name is dev.miku.r2dbc.mysql.extension.Extension, it contains this line:
org.example.demo.json.JacksonCodecRegistrar
Reporting Issues
The R2DBC MySQL Implementation uses GitHub as issue tracking system to record bugs and feature requests.
If you want to raise an issue, please follow the recommendations below:
Before log a bug, please search the issue tracker to see if someone has already reported the problem.
Please provide as much information as possible with the issue report, we like to know the version of R2DBC MySQL that you are using and JVM version.
If you need to paste code, or include a stack trace use Markdown ``` escapes before and after your text.
If possible try to create a test-case or project that replicates the issue. Attach a link to your code or a compressed file containing your code.
Before use
The MySQL data fields encoded by index-based natively, get fields by an index will have better performance than get by column name.
Each Result should be used (call getRowsUpdated or map, even table definition), can NOT just ignore any Result, otherwise inbound stream is unable to align. (like ResultSet.close in jdbc, Result auto-close after used by once)
The MySQL server does not actively return time zone when query DATETIME or TIMESTAMP, this driver does not attempt time zone conversion. That means should always use LocalDateTime for SQL type DATETIME or TIMESTAMP. Execute SHOW VARIABLES LIKE '%time_zone%' to get more information.
Should not turn-on the trace log level unless debugging. Otherwise, the security information may be exposed through ByteBuf dump.
If Statement bound returnGeneratedValues, the Result of the Statement can be called both: getRowsUpdated to get affected rows, and map to get last inserted ID.
The MySQL may be not support search rows by a binary field, like BIT, BLOB and JSON, because those data fields maybe just an address of reference in MySQL server, or maybe need strict bit-aligned. (but VARBINARY is OK)
License
This project is released under version 2.0 of the Apache License.
FAQs
R2DBC MySQL Implementation
We found that dev.miku:r2dbc-mysql demonstrated a not healthy version release cadence and project activity because the last version was released a year ago.It has 0 open source maintainers collaborating on the project.
Package last updated on 12 Jul 2020
Did you know?
Socket for GitHub automatically highlights issues in each pull request and monitors the health of all your open source dependencies. Discover the contents of your packages and block harmful activity before you install or update your dependencies.
Elastic’s return to open source with the AGPL license has been met with skepticism, as many developers see it as a strategic move rather than a genuine effort to restore user trust and freedoms.
A new "revival hijack" supply chain attack targets deleted Python packages, with an estimated 22K packages at risk. Socket can detect and block hijacked packages that have added malicious code.