diff --git a/docs/src/main/asciidoc/reactive-sql-clients.adoc b/docs/src/main/asciidoc/reactive-sql-clients.adoc index 05410696aba62..50685c3ccd535 100644 --- a/docs/src/main/asciidoc/reactive-sql-clients.adoc +++ b/docs/src/main/asciidoc/reactive-sql-clients.adoc @@ -688,6 +688,23 @@ In `application.properties` add: quarkus.datasource.reactive.url=mysql:///quarkus_test?host=/var/run/mysqld/mysqld.sock ---- +== Load-balancing connections + +The reactive PostgreSQL and MariaDB/MySQL clients support defining several connections. + +A typical configuration with several connections would look like: + +[source,properties] +---- +quarkus.datasource.reactive.url=postgresql://host1:5432/default,postgresql://host2:5432/default,postgresql://host3:5432/default +---- +This can also be written with indexed property syntax: +---- +quarkus.datasource.reactive.url[0]=postgresql://host1:5432/default +quarkus.datasource.reactive.url[1]=postgresql://host2:5432/default +quarkus.datasource.reactive.url[2]=postgresql://host3:5432/default +---- + == Pooled Connection `idle-timeout` Reactive datasources can be configured with an `idle-timeout` (in milliseconds). diff --git a/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/DataSourceReactiveRuntimeConfig.java b/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/DataSourceReactiveRuntimeConfig.java index c2e9de66a8f3f..7354c51d2ea48 100644 --- a/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/DataSourceReactiveRuntimeConfig.java +++ b/extensions/reactive-datasource/runtime/src/main/java/io/quarkus/reactive/datasource/runtime/DataSourceReactiveRuntimeConfig.java @@ -2,6 +2,7 @@ import java.time.Duration; import java.util.Collections; +import java.util.List; import java.util.Map; import java.util.Optional; import java.util.OptionalInt; @@ -23,10 +24,14 @@ public class DataSourceReactiveRuntimeConfig { public boolean cachePreparedStatements = false; /** - * The datasource URL. + * The datasource URLs. + *

+ * If multiple values are set, this datasource will create a pool with a list of servers instead of a single server. + * The pool uses a round-robin load balancing when a connection is created to select different servers. + * Note: some driver may not support multiple values here. */ @ConfigItem - public Optional url = Optional.empty(); + public Optional> url = Optional.empty(); /** * The datasource pool maximum size. diff --git a/extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java b/extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java index 8ac5b757336e8..6c04c5e312cb3 100644 --- a/extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java +++ b/extensions/reactive-db2-client/runtime/src/main/java/io/quarkus/reactive/db2/client/runtime/DB2PoolRecorder.java @@ -9,6 +9,7 @@ import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions; import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -119,7 +120,12 @@ private DB2ConnectOptions toConnectOptions(DataSourceRuntimeConfig dataSourceRun DB2ConnectOptions connectOptions; if (dataSourceReactiveRuntimeConfig.url.isPresent()) { - String url = dataSourceReactiveRuntimeConfig.url.get(); + List urls = dataSourceReactiveRuntimeConfig.url.get(); + if (urls.size() > 1) { + log.warn("The Reactive DB2 client does not support multiple URLs. The first one will be used, and " + + "others will be ignored."); + } + String url = urls.get(0); // clean up the URL to make migrations easier if (url.matches("^vertx-reactive:db2://.*$")) { url = url.substring("vertx-reactive:".length()); diff --git a/extensions/reactive-mssql-client/runtime/src/main/java/io/quarkus/reactive/mssql/client/runtime/MSSQLPoolRecorder.java b/extensions/reactive-mssql-client/runtime/src/main/java/io/quarkus/reactive/mssql/client/runtime/MSSQLPoolRecorder.java index 0d866b131a7c1..8a9073a040dd8 100644 --- a/extensions/reactive-mssql-client/runtime/src/main/java/io/quarkus/reactive/mssql/client/runtime/MSSQLPoolRecorder.java +++ b/extensions/reactive-mssql-client/runtime/src/main/java/io/quarkus/reactive/mssql/client/runtime/MSSQLPoolRecorder.java @@ -9,6 +9,7 @@ import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions; import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -118,7 +119,12 @@ private MSSQLConnectOptions toMSSQLConnectOptions(DataSourceRuntimeConfig dataSo DataSourceReactiveMSSQLConfig dataSourceReactiveMSSQLConfig) { MSSQLConnectOptions mssqlConnectOptions; if (dataSourceReactiveRuntimeConfig.url.isPresent()) { - String url = dataSourceReactiveRuntimeConfig.url.get(); + List urls = dataSourceReactiveRuntimeConfig.url.get(); + if (urls.size() > 1) { + log.warn("The Reactive MSSQL client does not support multiple URLs. The first one will be used, and " + + "others will be ignored."); + } + String url = urls.get(0); // clean up the URL to make migrations easier if (url.startsWith("vertx-reactive:sqlserver://")) { url = url.substring("vertx-reactive:".length()); diff --git a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/LocalhostMySQLPoolCreator.java b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/LocalhostMySQLPoolCreator.java index 71324a7b6d3b1..d8ec461a66b15 100644 --- a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/LocalhostMySQLPoolCreator.java +++ b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/LocalhostMySQLPoolCreator.java @@ -9,7 +9,7 @@ public class LocalhostMySQLPoolCreator implements MySQLPoolCreator { @Override public MySQLPool create(Input input) { - return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308), + return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308), input.poolOptions()); } } diff --git a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleDataSourcesAndMySQLPoolCreatorsTest.java b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleDataSourcesAndMySQLPoolCreatorsTest.java index 1ce79acd7e128..df0b1769391df 100644 --- a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleDataSourcesAndMySQLPoolCreatorsTest.java +++ b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleDataSourcesAndMySQLPoolCreatorsTest.java @@ -85,8 +85,8 @@ public static class DefaultMySQLPoolCreator implements MySQLPoolCreator { @Override public MySQLPool create(Input input) { - assertEquals(12345, input.mySQLConnectOptions().getPort()); // validate that the bean has been called for the proper datasource - return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308), + assertEquals(12345, input.mySQLConnectOptionsList().get(0).getPort()); // validate that the bean has been called for the proper datasource + return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308), input.poolOptions()); } } @@ -97,8 +97,8 @@ public static class HibernateMySQLPoolCreator implements MySQLPoolCreator { @Override public MySQLPool create(Input input) { - assertEquals(55555, input.mySQLConnectOptions().getPort()); // validate that the bean has been called for the proper datasource - return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions().setHost("localhost").setPort(3308), + assertEquals(55555, input.mySQLConnectOptionsList().get(0).getPort()); // validate that the bean has been called for the proper datasource + return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList().get(0).setHost("localhost").setPort(3308), input.poolOptions()); } } diff --git a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleMySQLPoolCreatorsForSameDatasourceTest.java b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleMySQLPoolCreatorsForSameDatasourceTest.java index 5016026dae1ea..d1a5ea845512c 100644 --- a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleMySQLPoolCreatorsForSameDatasourceTest.java +++ b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MultipleMySQLPoolCreatorsForSameDatasourceTest.java @@ -33,7 +33,7 @@ public static class AnotherMySQLPoolCreator implements MySQLPoolCreator { @Override public MySQLPool create(Input input) { - return MySQLPool.pool(input.vertx(), input.mySQLConnectOptions(), input.poolOptions()); + return MySQLPool.pool(input.vertx(), input.mySQLConnectOptionsList(), input.poolOptions()); } } diff --git a/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MySQLPoolLoadBalanceTest.java b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MySQLPoolLoadBalanceTest.java new file mode 100644 index 0000000000000..4f747ae2f08f9 --- /dev/null +++ b/extensions/reactive-mysql-client/deployment/src/test/java/io/quarkus/reactive/mysql/client/MySQLPoolLoadBalanceTest.java @@ -0,0 +1,31 @@ +package io.quarkus.reactive.mysql.client; + +import org.hamcrest.Matchers; +import org.jboss.shrinkwrap.api.asset.StringAsset; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.quarkus.test.QuarkusDevModeTest; +import io.restassured.RestAssured; + +public class MySQLPoolLoadBalanceTest { + + @RegisterExtension + public static final QuarkusDevModeTest test = new QuarkusDevModeTest() + .withApplicationRoot((jar) -> jar + .addClass(DevModeResource.class) + .add(new StringAsset("quarkus.datasource.db-kind=mysql\n" + + "quarkus.datasource.reactive.url=vertx-reactive:mysql://localhost:6033/load_balance_test," + + "vertx-reactive:mysql://localhost:6034/load_balance_test," + + "vertx-reactive:mysql://localhost:6035/load_balance_test"), + "application.properties")); + + @Test + public void testLoadBalance() { + RestAssured + .get("/dev/error") + .then() + .statusCode(200) + .body(Matchers.anyOf(Matchers.endsWith(":6033"), Matchers.endsWith(":6034"), Matchers.endsWith(":6035"))); + } +} diff --git a/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/MySQLPoolCreator.java b/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/MySQLPoolCreator.java index 43bfc3db99755..942ae82e32c35 100644 --- a/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/MySQLPoolCreator.java +++ b/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/MySQLPoolCreator.java @@ -1,5 +1,7 @@ package io.quarkus.reactive.mysql.client; +import java.util.List; + import io.quarkus.reactive.datasource.ReactiveDataSource; import io.vertx.core.Vertx; import io.vertx.mysqlclient.MySQLConnectOptions; @@ -25,6 +27,6 @@ interface Input { PoolOptions poolOptions(); - MySQLConnectOptions mySQLConnectOptions(); + List mySQLConnectOptionsList(); } } diff --git a/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/runtime/MySQLPoolRecorder.java b/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/runtime/MySQLPoolRecorder.java index d47125f2eaf9e..6c3e0ad654a38 100644 --- a/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/runtime/MySQLPoolRecorder.java +++ b/extensions/reactive-mysql-client/runtime/src/main/java/io/quarkus/reactive/mysql/client/runtime/MySQLPoolRecorder.java @@ -9,6 +9,8 @@ import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions; import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -72,16 +74,17 @@ private MySQLPool initialize(Vertx vertx, DataSourceReactiveMySQLConfig dataSourceReactiveMySQLConfig) { PoolOptions poolOptions = toPoolOptions(eventLoopCount, dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig, dataSourceReactiveMySQLConfig); - MySQLConnectOptions mysqlConnectOptions = toMySQLConnectOptions(dataSourceRuntimeConfig, + List mysqlConnectOptionsList = toMySQLConnectOptions(dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig, dataSourceReactiveMySQLConfig); // Use the convention defined by Quarkus Micrometer Vert.x metrics to create metrics prefixed with mysql. // and the client_name as tag. // See io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractPrefix and // io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractClientName - mysqlConnectOptions.setMetricsName("mysql|" + dataSourceName); + mysqlConnectOptionsList.forEach( + mysqlConnectOptions -> mysqlConnectOptions.setMetricsName("mysql|" + dataSourceName)); - return createPool(vertx, poolOptions, mysqlConnectOptions, dataSourceName); + return createPool(vertx, poolOptions, mysqlConnectOptionsList, dataSourceName); } private PoolOptions toPoolOptions(Integer eventLoopCount, @@ -119,103 +122,93 @@ private PoolOptions toPoolOptions(Integer eventLoopCount, return poolOptions; } - private MySQLConnectOptions toMySQLConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig, + private List toMySQLConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig, DataSourceReactiveRuntimeConfig dataSourceReactiveRuntimeConfig, DataSourceReactiveMySQLConfig dataSourceReactiveMySQLConfig) { - MySQLConnectOptions mysqlConnectOptions; + List mysqlConnectOptionsList = new ArrayList<>(); if (dataSourceReactiveRuntimeConfig.url.isPresent()) { - String url = dataSourceReactiveRuntimeConfig.url.get(); - // clean up the URL to make migrations easier - if (url.startsWith("vertx-reactive:mysql://")) { - url = url.substring("vertx-reactive:".length()); - } - mysqlConnectOptions = MySQLConnectOptions.fromUri(url); + List urls = dataSourceReactiveRuntimeConfig.url.get(); + urls.forEach(url -> { + // clean up the URL to make migrations easier + if (url.startsWith("vertx-reactive:mysql://")) { + url = url.substring("vertx-reactive:".length()); + } + mysqlConnectOptionsList.add(MySQLConnectOptions.fromUri(url)); + }); } else { - mysqlConnectOptions = new MySQLConnectOptions(); - } - - if (dataSourceRuntimeConfig.username.isPresent()) { - mysqlConnectOptions.setUser(dataSourceRuntimeConfig.username.get()); - } - - if (dataSourceRuntimeConfig.password.isPresent()) { - mysqlConnectOptions.setPassword(dataSourceRuntimeConfig.password.get()); - } - - // credentials provider - if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) { - String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null); - CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName); - String name = dataSourceRuntimeConfig.credentialsProvider.get(); - Map credentials = credentialsProvider.getCredentials(name); - String user = credentials.get(USER_PROPERTY_NAME); - String password = credentials.get(PASSWORD_PROPERTY_NAME); - if (user != null) { - mysqlConnectOptions.setUser(user); + mysqlConnectOptionsList.add(new MySQLConnectOptions()); + } + + mysqlConnectOptionsList.forEach(mysqlConnectOptions -> { + dataSourceRuntimeConfig.username.ifPresent(mysqlConnectOptions::setUser); + + dataSourceRuntimeConfig.password.ifPresent(mysqlConnectOptions::setPassword); + + // credentials provider + if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) { + String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null); + CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName); + String name = dataSourceRuntimeConfig.credentialsProvider.get(); + Map credentials = credentialsProvider.getCredentials(name); + String user = credentials.get(USER_PROPERTY_NAME); + String password = credentials.get(PASSWORD_PROPERTY_NAME); + if (user != null) { + mysqlConnectOptions.setUser(user); + } + if (password != null) { + mysqlConnectOptions.setPassword(password); + } } - if (password != null) { - mysqlConnectOptions.setPassword(password); - } - } - mysqlConnectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements); + mysqlConnectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements); - if (dataSourceReactiveMySQLConfig.charset.isPresent()) { - mysqlConnectOptions.setCharset(dataSourceReactiveMySQLConfig.charset.get()); - } - if (dataSourceReactiveMySQLConfig.collation.isPresent()) { - mysqlConnectOptions.setCollation(dataSourceReactiveMySQLConfig.collation.get()); - } + dataSourceReactiveMySQLConfig.charset.ifPresent(mysqlConnectOptions::setCharset); + dataSourceReactiveMySQLConfig.collation.ifPresent(mysqlConnectOptions::setCollation); - if (dataSourceReactiveMySQLConfig.pipeliningLimit.isPresent()) { - mysqlConnectOptions.setPipeliningLimit(dataSourceReactiveMySQLConfig.pipeliningLimit.getAsInt()); - } + if (dataSourceReactiveMySQLConfig.pipeliningLimit.isPresent()) { + mysqlConnectOptions.setPipeliningLimit(dataSourceReactiveMySQLConfig.pipeliningLimit.getAsInt()); + } - if (dataSourceReactiveMySQLConfig.useAffectedRows.isPresent()) { - mysqlConnectOptions.setUseAffectedRows(dataSourceReactiveMySQLConfig.useAffectedRows.get()); - } + dataSourceReactiveMySQLConfig.useAffectedRows.ifPresent(mysqlConnectOptions::setUseAffectedRows); - if (dataSourceReactiveMySQLConfig.sslMode.isPresent()) { - final SslMode sslMode = dataSourceReactiveMySQLConfig.sslMode.get(); - mysqlConnectOptions.setSslMode(sslMode); + if (dataSourceReactiveMySQLConfig.sslMode.isPresent()) { + final SslMode sslMode = dataSourceReactiveMySQLConfig.sslMode.get(); + mysqlConnectOptions.setSslMode(sslMode); - // If sslMode is verify-identity, we also need a hostname verification algorithm - if (sslMode == SslMode.VERIFY_IDENTITY && (!dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm - .isPresent() || "".equals(dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()))) { - throw new IllegalArgumentException( - "quarkus.datasource.reactive.hostname-verification-algorithm must be specified under verify-identity sslmode"); + // If sslMode is verify-identity, we also need a hostname verification algorithm + if (sslMode == SslMode.VERIFY_IDENTITY && (!dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm + .isPresent() || "".equals(dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()))) { + throw new IllegalArgumentException( + "quarkus.datasource.reactive.hostname-verification-algorithm must be specified under verify-identity sslmode"); + } } - } - mysqlConnectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll); + mysqlConnectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll); - configurePemTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem); - configureJksTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks); - configurePfxTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx); + configurePemTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem); + configureJksTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks); + configurePfxTrustOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx); - configurePemKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem); - configureJksKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks); - configurePfxKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx); + configurePemKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem); + configureJksKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks); + configurePfxKeyCertOptions(mysqlConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx); - mysqlConnectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts); + mysqlConnectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts); - mysqlConnectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis()); + mysqlConnectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis()); - if (dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.isPresent()) { - mysqlConnectOptions.setHostnameVerificationAlgorithm( - dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()); - } + dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.ifPresent( + mysqlConnectOptions::setHostnameVerificationAlgorithm); - if (dataSourceReactiveMySQLConfig.authenticationPlugin.isPresent()) { - mysqlConnectOptions.setAuthenticationPlugin(dataSourceReactiveMySQLConfig.authenticationPlugin.get()); - } + dataSourceReactiveMySQLConfig.authenticationPlugin.ifPresent(mysqlConnectOptions::setAuthenticationPlugin); - dataSourceReactiveRuntimeConfig.additionalProperties.forEach(mysqlConnectOptions::addProperty); + dataSourceReactiveRuntimeConfig.additionalProperties.forEach(mysqlConnectOptions::addProperty); + }); - return mysqlConnectOptions; + return mysqlConnectOptionsList; } - private MySQLPool createPool(Vertx vertx, PoolOptions poolOptions, MySQLConnectOptions mySQLConnectOptions, + private MySQLPool createPool(Vertx vertx, PoolOptions poolOptions, List mySQLConnectOptionsList, String dataSourceName) { Instance instance; if (DataSourceUtil.isDefault(dataSourceName)) { @@ -225,21 +218,21 @@ private MySQLPool createPool(Vertx vertx, PoolOptions poolOptions, MySQLConnectO new ReactiveDataSource.ReactiveDataSourceLiteral(dataSourceName)); } if (instance.isResolvable()) { - MySQLPoolCreator.Input input = new DefaultInput(vertx, poolOptions, mySQLConnectOptions); + MySQLPoolCreator.Input input = new DefaultInput(vertx, poolOptions, mySQLConnectOptionsList); return instance.get().create(input); } - return MySQLPool.pool(vertx, mySQLConnectOptions, poolOptions); + return MySQLPool.pool(vertx, mySQLConnectOptionsList, poolOptions); } private static class DefaultInput implements MySQLPoolCreator.Input { private final Vertx vertx; private final PoolOptions poolOptions; - private final MySQLConnectOptions mySQLConnectOptions; + private final List mySQLConnectOptionsList; - public DefaultInput(Vertx vertx, PoolOptions poolOptions, MySQLConnectOptions mySQLConnectOptions) { + public DefaultInput(Vertx vertx, PoolOptions poolOptions, List mySQLConnectOptionsList) { this.vertx = vertx; this.poolOptions = poolOptions; - this.mySQLConnectOptions = mySQLConnectOptions; + this.mySQLConnectOptionsList = mySQLConnectOptionsList; } @Override @@ -253,8 +246,8 @@ public PoolOptions poolOptions() { } @Override - public MySQLConnectOptions mySQLConnectOptions() { - return mySQLConnectOptions; + public List mySQLConnectOptionsList() { + return mySQLConnectOptionsList; } } } diff --git a/extensions/reactive-oracle-client/runtime/src/main/java/io/quarkus/reactive/oracle/client/runtime/OraclePoolRecorder.java b/extensions/reactive-oracle-client/runtime/src/main/java/io/quarkus/reactive/oracle/client/runtime/OraclePoolRecorder.java index 853f62aae1327..2c283f6839845 100644 --- a/extensions/reactive-oracle-client/runtime/src/main/java/io/quarkus/reactive/oracle/client/runtime/OraclePoolRecorder.java +++ b/extensions/reactive-oracle-client/runtime/src/main/java/io/quarkus/reactive/oracle/client/runtime/OraclePoolRecorder.java @@ -3,6 +3,7 @@ import static io.quarkus.credentials.CredentialsProvider.PASSWORD_PROPERTY_NAME; import static io.quarkus.credentials.CredentialsProvider.USER_PROPERTY_NAME; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -112,7 +113,12 @@ private OracleConnectOptions toOracleConnectOptions(DataSourceRuntimeConfig data DataSourceReactiveOracleConfig dataSourceReactiveOracleConfig) { OracleConnectOptions oracleConnectOptions; if (dataSourceReactiveRuntimeConfig.url.isPresent()) { - String url = dataSourceReactiveRuntimeConfig.url.get(); + List urls = dataSourceReactiveRuntimeConfig.url.get(); + if (urls.size() > 1) { + log.warn("The Reactive Oracle client does not support multiple URLs. The first one will be used, and " + + "others will be ignored."); + } + String url = urls.get(0); // clean up the URL to make migrations easier if (url.startsWith("vertx-reactive:oracle:")) { url = url.substring("vertx-reactive:".length()); diff --git a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/LocalhostPgPoolCreator.java b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/LocalhostPgPoolCreator.java index 79ec7eb2890b1..935202ff76a73 100644 --- a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/LocalhostPgPoolCreator.java +++ b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/LocalhostPgPoolCreator.java @@ -9,6 +9,7 @@ public class LocalhostPgPoolCreator implements PgPoolCreator { @Override public PgPool create(Input input) { - return PgPool.pool(input.vertx(), input.pgConnectOptions().setHost("localhost").setPort(5431), input.poolOptions()); + return PgPool.pool(input.vertx(), input.pgConnectOptionsList().get(0).setHost("localhost").setPort(5431), + input.poolOptions()); } } diff --git a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultipleDataSourcesAndPgPoolCreatorsTest.java b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultipleDataSourcesAndPgPoolCreatorsTest.java index ccd0e4ff09be7..bf1395094ccbb 100644 --- a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultipleDataSourcesAndPgPoolCreatorsTest.java +++ b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultipleDataSourcesAndPgPoolCreatorsTest.java @@ -85,8 +85,9 @@ public static class DefaultPgPoolCreator implements PgPoolCreator { @Override public PgPool create(Input input) { - assertEquals(10, input.pgConnectOptions().getPipeliningLimit()); // validate that the bean has been called for the proper datasource - return PgPool.pool(input.vertx(), input.pgConnectOptions().setHost("localhost").setPort(5431), input.poolOptions()); + assertEquals(10, input.pgConnectOptionsList().get(0).getPipeliningLimit()); // validate that the bean has been called for the proper datasource + return PgPool.pool(input.vertx(), input.pgConnectOptionsList().get(0).setHost("localhost").setPort(5431), + input.poolOptions()); } } @@ -96,8 +97,9 @@ public static class HibernatePgPoolCreator implements PgPoolCreator { @Override public PgPool create(Input input) { - assertEquals(7, input.pgConnectOptions().getPipeliningLimit()); // validate that the bean has been called for the proper datasource - return PgPool.pool(input.vertx(), input.pgConnectOptions().setHost("localhost").setPort(5431), input.poolOptions()); + assertEquals(7, input.pgConnectOptionsList().get(0).getPipeliningLimit()); // validate that the bean has been called for the proper datasource + return PgPool.pool(input.vertx(), input.pgConnectOptionsList().get(0).setHost("localhost").setPort(5431), + input.poolOptions()); } } } diff --git a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultiplePgPoolCreatorsForSameDatasourceTest.java b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultiplePgPoolCreatorsForSameDatasourceTest.java index ea8b3a5e5b30e..f605bba13b9b3 100644 --- a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultiplePgPoolCreatorsForSameDatasourceTest.java +++ b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/MultiplePgPoolCreatorsForSameDatasourceTest.java @@ -33,7 +33,7 @@ public static class AnotherPgPoolCreator implements PgPoolCreator { @Override public PgPool create(Input input) { - return PgPool.pool(input.vertx(), input.pgConnectOptions(), input.poolOptions()); + return PgPool.pool(input.vertx(), input.pgConnectOptionsList(), input.poolOptions()); } } diff --git a/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/PgPoolLoadBalanceTest.java b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/PgPoolLoadBalanceTest.java new file mode 100644 index 0000000000000..bfb1b5165ad3a --- /dev/null +++ b/extensions/reactive-pg-client/deployment/src/test/java/io/quarkus/reactive/pg/client/PgPoolLoadBalanceTest.java @@ -0,0 +1,31 @@ +package io.quarkus.reactive.pg.client; + +import org.hamcrest.Matchers; +import org.jboss.shrinkwrap.api.asset.StringAsset; +import org.junit.jupiter.api.Test; +import org.junit.jupiter.api.extension.RegisterExtension; + +import io.quarkus.test.QuarkusDevModeTest; +import io.restassured.RestAssured; + +public class PgPoolLoadBalanceTest { + + @RegisterExtension + public static final QuarkusDevModeTest test = new QuarkusDevModeTest() + .withApplicationRoot((jar) -> jar + .addClass(DevModeResource.class) + .add(new StringAsset("quarkus.datasource.db-kind=postgresql\n" + + "quarkus.datasource.reactive.url=vertx-reactive:postgresql://localhost:5342/load_balance_test," + + "vertx-reactive:postgresql://localhost:5343/load_balance_test," + + "vertx-reactive:postgresql://localhost:5344/load_balance_test"), + "application.properties")); + + @Test + public void testLoadBalance() { + RestAssured + .get("/dev/error") + .then() + .statusCode(200) + .body(Matchers.anyOf(Matchers.endsWith(":5342"), Matchers.endsWith(":5343"), Matchers.endsWith(":5344"))); + } +} diff --git a/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/PgPoolCreator.java b/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/PgPoolCreator.java index ca40189b97b6e..34bb67c732742 100644 --- a/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/PgPoolCreator.java +++ b/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/PgPoolCreator.java @@ -1,5 +1,7 @@ package io.quarkus.reactive.pg.client; +import java.util.List; + import io.quarkus.reactive.datasource.ReactiveDataSource; import io.vertx.core.Vertx; import io.vertx.pgclient.PgConnectOptions; @@ -25,6 +27,6 @@ interface Input { PoolOptions poolOptions(); - PgConnectOptions pgConnectOptions(); + List pgConnectOptionsList(); } } diff --git a/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/runtime/PgPoolRecorder.java b/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/runtime/PgPoolRecorder.java index 79088c5bcf2b7..d7763a6d5f4e2 100644 --- a/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/runtime/PgPoolRecorder.java +++ b/extensions/reactive-pg-client/runtime/src/main/java/io/quarkus/reactive/pg/client/runtime/PgPoolRecorder.java @@ -9,6 +9,8 @@ import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxKeyCertOptions; import static io.quarkus.vertx.core.runtime.SSLConfigHelper.configurePfxTrustOptions; +import java.util.ArrayList; +import java.util.List; import java.util.Map; import java.util.concurrent.TimeUnit; import java.util.function.Supplier; @@ -73,15 +75,15 @@ private PgPool initialize(Vertx vertx, DataSourceReactivePostgreSQLConfig dataSourceReactivePostgreSQLConfig) { PoolOptions poolOptions = toPoolOptions(eventLoopCount, dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig, dataSourceReactivePostgreSQLConfig); - PgConnectOptions pgConnectOptions = toPgConnectOptions(dataSourceRuntimeConfig, dataSourceReactiveRuntimeConfig, - dataSourceReactivePostgreSQLConfig); + List pgConnectOptionsList = toPgConnectOptions(dataSourceRuntimeConfig, + dataSourceReactiveRuntimeConfig, dataSourceReactivePostgreSQLConfig); // Use the convention defined by Quarkus Micrometer Vert.x metrics to create metrics prefixed with postgresql. // and the client_name as tag. // See io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractPrefix and // io.quarkus.micrometer.runtime.binder.vertx.VertxMeterBinderAdapter.extractClientName - pgConnectOptions.setMetricsName("postgresql|" + dataSourceName); + pgConnectOptionsList.forEach(pgConnectOptions -> pgConnectOptions.setMetricsName("postgresql|" + dataSourceName)); - return createPool(vertx, poolOptions, pgConnectOptions, dataSourceName); + return createPool(vertx, poolOptions, pgConnectOptionsList, dataSourceName); } private PoolOptions toPoolOptions(Integer eventLoopCount, @@ -114,89 +116,87 @@ private PoolOptions toPoolOptions(Integer eventLoopCount, return poolOptions; } - private PgConnectOptions toPgConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig, + private List toPgConnectOptions(DataSourceRuntimeConfig dataSourceRuntimeConfig, DataSourceReactiveRuntimeConfig dataSourceReactiveRuntimeConfig, DataSourceReactivePostgreSQLConfig dataSourceReactivePostgreSQLConfig) { - PgConnectOptions pgConnectOptions; + List pgConnectOptionsList = new ArrayList<>(); if (dataSourceReactiveRuntimeConfig.url.isPresent()) { - String url = dataSourceReactiveRuntimeConfig.url.get(); - // clean up the URL to make migrations easier - if (url.matches("^vertx-reactive:postgre(?:s|sql)://.*$")) { - url = url.substring("vertx-reactive:".length()); - } - pgConnectOptions = PgConnectOptions.fromUri(url); + List urls = dataSourceReactiveRuntimeConfig.url.get(); + urls.forEach(url -> { + // clean up the URL to make migrations easier + if (url.matches("^vertx-reactive:postgre(?:s|sql)://.*$")) { + url = url.substring("vertx-reactive:".length()); + } + pgConnectOptionsList.add(PgConnectOptions.fromUri(url)); + }); } else { - pgConnectOptions = new PgConnectOptions(); - } - - if (dataSourceRuntimeConfig.username.isPresent()) { - pgConnectOptions.setUser(dataSourceRuntimeConfig.username.get()); - } - - if (dataSourceRuntimeConfig.password.isPresent()) { - pgConnectOptions.setPassword(dataSourceRuntimeConfig.password.get()); - } - - // credentials provider - if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) { - String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null); - CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName); - String name = dataSourceRuntimeConfig.credentialsProvider.get(); - Map credentials = credentialsProvider.getCredentials(name); - String user = credentials.get(USER_PROPERTY_NAME); - String password = credentials.get(PASSWORD_PROPERTY_NAME); - if (user != null) { - pgConnectOptions.setUser(user); - } - if (password != null) { - pgConnectOptions.setPassword(password); + pgConnectOptionsList.add(new PgConnectOptions()); + } + + pgConnectOptionsList.forEach(pgConnectOptions -> { + dataSourceRuntimeConfig.username.ifPresent(pgConnectOptions::setUser); + + dataSourceRuntimeConfig.password.ifPresent(pgConnectOptions::setPassword); + + // credentials provider + if (dataSourceRuntimeConfig.credentialsProvider.isPresent()) { + String beanName = dataSourceRuntimeConfig.credentialsProviderName.orElse(null); + CredentialsProvider credentialsProvider = CredentialsProviderFinder.find(beanName); + String name = dataSourceRuntimeConfig.credentialsProvider.get(); + Map credentials = credentialsProvider.getCredentials(name); + String user = credentials.get(USER_PROPERTY_NAME); + String password = credentials.get(PASSWORD_PROPERTY_NAME); + if (user != null) { + pgConnectOptions.setUser(user); + } + if (password != null) { + pgConnectOptions.setPassword(password); + } } - } - pgConnectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements); + pgConnectOptions.setCachePreparedStatements(dataSourceReactiveRuntimeConfig.cachePreparedStatements); - if (dataSourceReactivePostgreSQLConfig.pipeliningLimit.isPresent()) { - pgConnectOptions.setPipeliningLimit(dataSourceReactivePostgreSQLConfig.pipeliningLimit.getAsInt()); - } + if (dataSourceReactivePostgreSQLConfig.pipeliningLimit.isPresent()) { + pgConnectOptions.setPipeliningLimit(dataSourceReactivePostgreSQLConfig.pipeliningLimit.getAsInt()); + } - if (dataSourceReactivePostgreSQLConfig.sslMode.isPresent()) { - final SslMode sslMode = dataSourceReactivePostgreSQLConfig.sslMode.get(); - pgConnectOptions.setSslMode(sslMode); + if (dataSourceReactivePostgreSQLConfig.sslMode.isPresent()) { + final SslMode sslMode = dataSourceReactivePostgreSQLConfig.sslMode.get(); + pgConnectOptions.setSslMode(sslMode); - // If sslMode is verify-full, we also need a hostname verification algorithm - if (sslMode == SslMode.VERIFY_FULL && (!dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm - .isPresent() || "".equals(dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()))) { - throw new IllegalArgumentException( - "quarkus.datasource.reactive.hostname-verification-algorithm must be specified under verify-full sslmode"); + // If sslMode is verify-full, we also need a hostname verification algorithm + if (sslMode == SslMode.VERIFY_FULL && (!dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm + .isPresent() || "".equals(dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()))) { + throw new IllegalArgumentException( + "quarkus.datasource.reactive.hostname-verification-algorithm must be specified under verify-full sslmode"); + } } - } - pgConnectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll); + pgConnectOptions.setTrustAll(dataSourceReactiveRuntimeConfig.trustAll); - configurePemTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem); - configureJksTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks); - configurePfxTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx); + configurePemTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePem); + configureJksTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificateJks); + configurePfxTrustOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.trustCertificatePfx); - configurePemKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem); - configureJksKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks); - configurePfxKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx); + configurePemKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePem); + configureJksKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificateJks); + configurePfxKeyCertOptions(pgConnectOptions, dataSourceReactiveRuntimeConfig.keyCertificatePfx); - pgConnectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts); + pgConnectOptions.setReconnectAttempts(dataSourceReactiveRuntimeConfig.reconnectAttempts); - pgConnectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis()); + pgConnectOptions.setReconnectInterval(dataSourceReactiveRuntimeConfig.reconnectInterval.toMillis()); - if (dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.isPresent()) { - pgConnectOptions.setHostnameVerificationAlgorithm( - dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.get()); - } + dataSourceReactiveRuntimeConfig.hostnameVerificationAlgorithm.ifPresent( + pgConnectOptions::setHostnameVerificationAlgorithm); - dataSourceReactiveRuntimeConfig.additionalProperties.forEach(pgConnectOptions::addProperty); + dataSourceReactiveRuntimeConfig.additionalProperties.forEach(pgConnectOptions::addProperty); + }); - return pgConnectOptions; + return pgConnectOptionsList; } - private PgPool createPool(Vertx vertx, PoolOptions poolOptions, PgConnectOptions pgConnectOptions, + private PgPool createPool(Vertx vertx, PoolOptions poolOptions, List pgConnectOptionsList, String dataSourceName) { Instance instance; if (DataSourceUtil.isDefault(dataSourceName)) { @@ -206,21 +206,21 @@ private PgPool createPool(Vertx vertx, PoolOptions poolOptions, PgConnectOptions new ReactiveDataSource.ReactiveDataSourceLiteral(dataSourceName)); } if (instance.isResolvable()) { - PgPoolCreator.Input input = new DefaultInput(vertx, poolOptions, pgConnectOptions); + PgPoolCreator.Input input = new DefaultInput(vertx, poolOptions, pgConnectOptionsList); return instance.get().create(input); } - return PgPool.pool(vertx, pgConnectOptions, poolOptions); + return PgPool.pool(vertx, pgConnectOptionsList, poolOptions); } private static class DefaultInput implements PgPoolCreator.Input { private final Vertx vertx; private final PoolOptions poolOptions; - private final PgConnectOptions pgConnectOptions; + private final List pgConnectOptionsList; - public DefaultInput(Vertx vertx, PoolOptions poolOptions, PgConnectOptions pgConnectOptions) { + public DefaultInput(Vertx vertx, PoolOptions poolOptions, List pgConnectOptionsList) { this.vertx = vertx; this.poolOptions = poolOptions; - this.pgConnectOptions = pgConnectOptions; + this.pgConnectOptionsList = pgConnectOptionsList; } @Override @@ -234,8 +234,8 @@ public PoolOptions poolOptions() { } @Override - public PgConnectOptions pgConnectOptions() { - return pgConnectOptions; + public List pgConnectOptionsList() { + return pgConnectOptionsList; } } }