resolvedConnectionFactories;
+
+ private @Nullable ConnectionFactory resolvedDefaultConnectionFactory;
+
+ /**
+ * Specify the map of target {@link ConnectionFactory ConnectionFactories}, with the lookup key as key. The mapped
+ * value can either be a corresponding {@link ConnectionFactory} instance or a connection factory name String (to be
+ * resolved via a {@link #setConnectionFactoryLookup ConnectionFactoryLookup}).
+ *
+ * The key can be of arbitrary type; this class implements the generic lookup process only. The concrete key
+ * representation will be handled by {@link #resolveSpecifiedLookupKey(Object)} and
+ * {@link #determineCurrentLookupKey()}.
+ */
+ @SuppressWarnings("unchecked")
+ public void setTargetConnectionFactories(Map targetConnectionFactories) {
+ this.targetConnectionFactories = (Map) targetConnectionFactories;
+ }
+
+ /**
+ * Specify the default target {@link ConnectionFactory}, if any.
+ *
+ * The mapped value can either be a corresponding {@link ConnectionFactory} instance or a connection factory name
+ * {@link String} (to be resolved via a {@link #setConnectionFactoryLookup ConnectionFactoryLookup}).
+ *
+ * This {@link ConnectionFactory} will be used as target if none of the keyed {@link #setTargetConnectionFactories
+ * targetConnectionFactories} match the {@link #determineCurrentLookupKey()} current lookup key.
+ */
+ public void setDefaultTargetConnectionFactory(Object defaultTargetConnectionFactory) {
+ this.defaultTargetConnectionFactory = defaultTargetConnectionFactory;
+ }
+
+ /**
+ * Specify whether to apply a lenient fallback to the default {@link ConnectionFactory} if no specific
+ * {@link ConnectionFactory} could be found for the current lookup key.
+ *
+ * Default is {@literal true}, accepting lookup keys without a corresponding entry in the target
+ * {@link ConnectionFactory} map - simply falling back to the default {@link ConnectionFactory} in that case.
+ *
+ * Switch this flag to {@literal false} if you would prefer the fallback to only apply no lookup key was emitted.
+ * Lookup keys without a {@link ConnectionFactory} entry will then lead to an {@link IllegalStateException}.
+ *
+ * @see #setTargetConnectionFactories
+ * @see #setDefaultTargetConnectionFactory
+ * @see #determineCurrentLookupKey()
+ */
+ public void setLenientFallback(boolean lenientFallback) {
+ this.lenientFallback = lenientFallback;
+ }
+
+ /**
+ * Set the {@link ConnectionFactoryLookup} implementation to use for resolving connection factory name Strings in the
+ * {@link #setTargetConnectionFactories targetConnectionFactories} map.
+ */
+ public void setConnectionFactoryLookup(ConnectionFactoryLookup connectionFactoryLookup) {
+
+ Assert.notNull(connectionFactoryLookup, "ConnectionFactoryLookup must not be null!");
+
+ this.connectionFactoryLookup = connectionFactoryLookup;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.beans.factory.InitializingBean#afterPropertiesSet()
+ */
+ @Override
+ public void afterPropertiesSet() {
+
+ Assert.notNull(this.targetConnectionFactories, "Property 'targetConnectionFactories' must not be null!");
+
+ this.resolvedConnectionFactories = new HashMap<>(this.targetConnectionFactories.size());
+ this.targetConnectionFactories.forEach((key, value) -> {
+ Object lookupKey = resolveSpecifiedLookupKey(key);
+ ConnectionFactory connectionFactory = resolveSpecifiedConnectionFactory(value);
+ this.resolvedConnectionFactories.put(lookupKey, connectionFactory);
+ });
+
+ if (this.defaultTargetConnectionFactory != null) {
+ this.resolvedDefaultConnectionFactory = resolveSpecifiedConnectionFactory(this.defaultTargetConnectionFactory);
+ }
+ }
+
+ /**
+ * Resolve the given lookup key object, as specified in the {@link #setTargetConnectionFactories
+ * targetConnectionFactories} map, into the actual lookup key to be used for matching with the
+ * {@link #determineCurrentLookupKey() current lookup key}.
+ *
+ * The default implementation simply returns the given key as-is.
+ *
+ * @param lookupKey the lookup key object as specified by the user.
+ * @return the lookup key as needed for matching.
+ */
+ protected Object resolveSpecifiedLookupKey(Object lookupKey) {
+ return lookupKey;
+ }
+
+ /**
+ * Resolve the specified connection factory object into a {@link ConnectionFactory} instance.
+ *
+ * The default implementation handles {@link ConnectionFactory} instances and connection factory names (to be resolved
+ * via a {@link #setConnectionFactoryLookup ConnectionFactoryLookup}).
+ *
+ * @param connectionFactory the connection factory value object as specified in the
+ * {@link #setTargetConnectionFactories targetConnectionFactories} map.
+ * @return the resolved {@link ConnectionFactory} (never {@literal null}).
+ * @throws IllegalArgumentException in case of an unsupported value type.
+ */
+ protected ConnectionFactory resolveSpecifiedConnectionFactory(Object connectionFactory)
+ throws IllegalArgumentException {
+
+ if (connectionFactory instanceof ConnectionFactory) {
+ return (ConnectionFactory) connectionFactory;
+ } else if (connectionFactory instanceof String) {
+ return this.connectionFactoryLookup.getConnectionFactory((String) connectionFactory);
+ } else {
+ throw new IllegalArgumentException(
+ "Illegal connection factory value - only 'io.r2dbc.spi.ConnectionFactory' and 'String' supported: "
+ + connectionFactory);
+ }
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see io.r2dbc.spi.ConnectionFactory#create()
+ */
+ @Override
+ public Mono create() {
+
+ return determineTargetConnectionFactory() //
+ .map(ConnectionFactory::create) //
+ .flatMap(Mono::from);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see io.r2dbc.spi.ConnectionFactory#getMetadata()
+ */
+ @Override
+ public ConnectionFactoryMetadata getMetadata() {
+
+ if (this.resolvedDefaultConnectionFactory != null) {
+ return this.resolvedDefaultConnectionFactory.getMetadata();
+ }
+
+ throw new UnsupportedOperationException(
+ "No default ConnectionFactory configured to retrieve ConnectionFactoryMetadata");
+ }
+
+ /**
+ * Retrieve the current target {@link ConnectionFactory}. Determines the {@link #determineCurrentLookupKey() current
+ * lookup key}, performs a lookup in the {@link #setTargetConnectionFactories targetConnectionFactories} map, falls
+ * back to the specified {@link #setDefaultTargetConnectionFactory default target ConnectionFactory} if necessary.
+ *
+ * @see #determineCurrentLookupKey()
+ * @return {@link Mono} emitting the current {@link ConnectionFactory} as per {@link #determineCurrentLookupKey()}.
+ */
+ protected Mono determineTargetConnectionFactory() {
+
+ Assert.state(this.resolvedConnectionFactories != null, "ConnectionFactory router not initialized");
+
+ Mono lookupKey = determineCurrentLookupKey().defaultIfEmpty(FALLBACK_MARKER);
+
+ return lookupKey.handle((key, sink) -> {
+
+ ConnectionFactory connectionFactory = this.resolvedConnectionFactories.get(key);
+
+ if (connectionFactory == null && (key == FALLBACK_MARKER || this.lenientFallback)) {
+ connectionFactory = this.resolvedDefaultConnectionFactory;
+ }
+
+ if (connectionFactory == null) {
+ sink.error(new IllegalStateException(String.format(
+ "Cannot determine target ConnectionFactory for lookup key '%s'", key == FALLBACK_MARKER ? null : key)));
+ return;
+ }
+
+ sink.next(connectionFactory);
+ });
+ }
+
+ /**
+ * Determine the current lookup key. This will typically be implemented to check a subscriber context. Allows for
+ * arbitrary keys. The returned key needs to match the stored lookup key type, as resolved by the
+ * {@link #resolveSpecifiedLookupKey} method.
+ *
+ * @return {@link Mono} emitting the lookup key. May complete without emitting a value if no lookup key available.
+ */
+ protected abstract Mono determineCurrentLookupKey();
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookup.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookup.java
new file mode 100644
index 00000000..9f824050
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookup.java
@@ -0,0 +1,90 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.lang.Nullable;
+import org.springframework.util.Assert;
+
+/**
+ * {@link ConnectionFactoryLookup} implementation based on a Spring {@link BeanFactory}.
+ *
+ * Will lookup Spring managed beans identified by bean name, expecting them to be of type {@link ConnectionFactory}.
+ *
+ * @author Mark Paluch
+ * @see BeanFactory
+ */
+public class BeanFactoryConnectionFactoryLookup implements ConnectionFactoryLookup, BeanFactoryAware {
+
+ @Nullable private BeanFactory beanFactory;
+
+ /**
+ * Creates a new {@link BeanFactoryConnectionFactoryLookup} instance.
+ *
+ * The {@link BeanFactory} to access must be set via {@code setBeanFactory}.
+ *
+ * @see #setBeanFactory
+ */
+ public BeanFactoryConnectionFactoryLookup() {}
+
+ /**
+ * Create a new instance of the {@link BeanFactoryConnectionFactoryLookup} class.
+ *
+ * Use of this constructor is redundant if this object is being created by a Spring IoC container, as the supplied
+ * {@link BeanFactory} will be replaced by the {@link BeanFactory} that creates it (see the {@link BeanFactoryAware}
+ * contract). So only use this constructor if you are using this class outside the context of a Spring IoC container.
+ *
+ * @param beanFactory the bean factory to be used to lookup {@link ConnectionFactory ConnectionFactories}.
+ */
+ public BeanFactoryConnectionFactoryLookup(BeanFactory beanFactory) {
+
+ Assert.notNull(beanFactory, "BeanFactory must not be null!");
+
+ this.beanFactory = beanFactory;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.beans.factory.BeanFactoryAware#setBeanFactory(org.springframework.beans.factory.BeanFactory)
+ */
+ @Override
+ public void setBeanFactory(BeanFactory beanFactory) {
+ this.beanFactory = beanFactory;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.connectionfactory.lookup.ConnectionFactoryLookup#getConnectionFactory(java.lang.String)
+ */
+ @Override
+ public ConnectionFactory getConnectionFactory(String connectionFactoryName)
+ throws ConnectionFactoryLookupFailureException {
+
+ Assert.state(this.beanFactory != null, "BeanFactory must not be null!");
+
+ try {
+ return this.beanFactory.getBean(connectionFactoryName, ConnectionFactory.class);
+ } catch (BeansException ex) {
+ throw new ConnectionFactoryLookupFailureException(
+ String.format("Failed to look up ConnectionFactory bean with name '%s'", connectionFactoryName), ex);
+ }
+ }
+
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookup.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookup.java
new file mode 100644
index 00000000..f8f76225
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookup.java
@@ -0,0 +1,36 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+/**
+ * Strategy interface for looking up {@link ConnectionFactory} by name.
+ *
+ * @author Mark Paluch
+ */
+@FunctionalInterface
+public interface ConnectionFactoryLookup {
+
+ /**
+ * Retrieve the {@link ConnectionFactory} identified by the given name.
+ *
+ * @param connectionFactoryName the name of the {@link ConnectionFactory}.
+ * @return the {@link ConnectionFactory} (never {@literal null}).
+ * @throws ConnectionFactoryLookupFailureException if the lookup failed.
+ */
+ ConnectionFactory getConnectionFactory(String connectionFactoryName) throws ConnectionFactoryLookupFailureException;
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookupFailureException.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookupFailureException.java
new file mode 100644
index 00000000..11889aca
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookupFailureException.java
@@ -0,0 +1,47 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import org.springframework.dao.NonTransientDataAccessException;
+
+/**
+ * Exception to be thrown by a {@link ConnectionFactoryLookup} implementation, indicating that the specified
+ * {@link io.r2dbc.spi.ConnectionFactory} could not be obtained.
+ *
+ * @author Mark Paluch
+ */
+@SuppressWarnings("serial")
+public class ConnectionFactoryLookupFailureException extends NonTransientDataAccessException {
+
+ /**
+ * Constructor for {@link ConnectionFactoryLookupFailureException}.
+ *
+ * @param msg the detail message.
+ */
+ public ConnectionFactoryLookupFailureException(String msg) {
+ super(msg);
+ }
+
+ /**
+ * Constructor for {@link ConnectionFactoryLookupFailureException}.
+ *
+ * @param msg the detail message.
+ * @param cause the root cause.
+ */
+ public ConnectionFactoryLookupFailureException(String msg, Throwable cause) {
+ super(msg, cause);
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookup.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookup.java
new file mode 100644
index 00000000..10a17ef3
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookup.java
@@ -0,0 +1,123 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.util.Assert;
+
+/**
+ * Simple {@link ConnectionFactoryLookup} implementation that relies on a map for doing lookups.
+ *
+ * Useful for testing environments or applications that need to match arbitrary {@link String} names to target
+ * {@link ConnectionFactory} objects.
+ *
+ * @author Mark Paluch
+ */
+public class MapConnectionFactoryLookup implements ConnectionFactoryLookup {
+
+ private final Map connectionFactories = new HashMap<>();
+
+ /**
+ * Create a new instance of the {@link MapConnectionFactoryLookup} class.
+ */
+ public MapConnectionFactoryLookup() {}
+
+ /**
+ * Create a new instance of the {@link MapConnectionFactoryLookup} class.
+ *
+ * @param connectionFactories the {@link Map} of {@link ConnectionFactory}. The keys are {@link String Strings}, the
+ * values are actual {@link ConnectionFactory} instances.
+ */
+ public MapConnectionFactoryLookup(Map connectionFactories) {
+ setConnectionFactories(connectionFactories);
+ }
+
+ /**
+ * Create a new instance of the {@link MapConnectionFactoryLookup} class.
+ *
+ * @param connectionFactoryName the name under which the supplied {@link ConnectionFactory} is to be added
+ * @param connectionFactory the {@link ConnectionFactory} to be added
+ */
+ public MapConnectionFactoryLookup(String connectionFactoryName, ConnectionFactory connectionFactory) {
+ addConnectionFactory(connectionFactoryName, connectionFactory);
+ }
+
+ /**
+ * Set the {@link Map} of {@link ConnectionFactory ConnectionFactories}. The keys are {@link String Strings}, the
+ * values are actual {@link ConnectionFactory} instances.
+ *
+ * If the supplied {@link Map} is {@code null}, then this method call effectively has no effect.
+ *
+ * @param connectionFactories said {@link Map} of {@link ConnectionFactory connectionFactories}
+ */
+ public void setConnectionFactories(Map connectionFactories) {
+
+ Assert.notNull(connectionFactories, "ConnectionFactories must not be null!");
+
+ this.connectionFactories.putAll(connectionFactories);
+ }
+
+ /**
+ * Get the {@link Map} of {@link ConnectionFactory ConnectionFactories} maintained by this object.
+ *
+ * The returned {@link Map} is {@link Collections#unmodifiableMap(Map) unmodifiable}.
+ *
+ * @return {@link Map} of {@link ConnectionFactory connectionFactory} (never {@literal null}).
+ */
+ public Map getConnectionFactories() {
+ return Collections.unmodifiableMap(this.connectionFactories);
+ }
+
+ /**
+ * Add the supplied {@link ConnectionFactory} to the map of {@link ConnectionFactory ConnectionFactorys} maintained by
+ * this object.
+ *
+ * @param connectionFactoryName the name under which the supplied {@link ConnectionFactory} is to be added
+ * @param connectionFactory the {@link ConnectionFactory} to be so added
+ */
+ public void addConnectionFactory(String connectionFactoryName, ConnectionFactory connectionFactory) {
+
+ Assert.notNull(connectionFactoryName, "ConnectionFactory name must not be null!");
+ Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
+
+ this.connectionFactories.put(connectionFactoryName, connectionFactory);
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.connectionfactory.lookup.ConnectionFactoryLookup#getConnectionFactory(java.lang.String)
+ */
+ @Override
+ public ConnectionFactory getConnectionFactory(String connectionFactoryName)
+ throws ConnectionFactoryLookupFailureException {
+
+ Assert.notNull(connectionFactoryName, "ConnectionFactory name must not be null!");
+
+ ConnectionFactory connectionFactory = this.connectionFactories.get(connectionFactoryName);
+
+ if (connectionFactory == null) {
+ throw new ConnectionFactoryLookupFailureException(
+ "No ConnectionFactory with name '" + connectionFactoryName + "' registered");
+ }
+
+ return connectionFactory;
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/SingleConnectionFactoryLookup.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/SingleConnectionFactoryLookup.java
new file mode 100644
index 00000000..85c7f986
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/SingleConnectionFactoryLookup.java
@@ -0,0 +1,53 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import org.springframework.util.Assert;
+
+/**
+ * An implementation of {@link ConnectionFactoryLookup} that simply wraps a single given {@link ConnectionFactory},
+ * returned for any connection factory name.
+ *
+ * @author Mark Paluch
+ */
+public class SingleConnectionFactoryLookup implements ConnectionFactoryLookup {
+
+ private final ConnectionFactory connectionFactory;
+
+ /**
+ * Create a new instance of the {@link SingleConnectionFactoryLookup} class.
+ *
+ * @param connectionFactory the single {@link ConnectionFactory} to wrap.
+ */
+ public SingleConnectionFactoryLookup(ConnectionFactory connectionFactory) {
+
+ Assert.notNull(connectionFactory, "ConnectionFactory must not be null!");
+
+ this.connectionFactory = connectionFactory;
+ }
+
+ /*
+ * (non-Javadoc)
+ * @see org.springframework.data.r2dbc.connectionfactory.lookup.ConnectionFactoryLookup#getConnectionFactory(java.lang.String)
+ */
+ @Override
+ public ConnectionFactory getConnectionFactory(String connectionFactoryName)
+ throws ConnectionFactoryLookupFailureException {
+ return this.connectionFactory;
+ }
+}
diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/package-info.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/package-info.java
new file mode 100644
index 00000000..efb4e618
--- /dev/null
+++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/package-info.java
@@ -0,0 +1,6 @@
+/**
+ * Provides a strategy for looking up R2DBC ConnectionFactories by name.
+ */
+@org.springframework.lang.NonNullApi
+@org.springframework.lang.NonNullFields
+package org.springframework.data.r2dbc.connectionfactory.lookup;
diff --git a/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactoryUnitTests.java b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactoryUnitTests.java
new file mode 100644
index 00000000..2feda2ca
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactoryUnitTests.java
@@ -0,0 +1,191 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+
+import io.r2dbc.spi.ConnectionFactory;
+import reactor.core.publisher.Mono;
+import reactor.test.StepVerifier;
+import reactor.util.context.Context;
+
+import java.util.Collections;
+import java.util.Map;
+
+import org.junit.Before;
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+/**
+ * Unit tests for {@link AbstractRoutingConnectionFactory}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class AbstractRoutingConnectionFactoryUnitTests {
+
+ private static final String ROUTING_KEY = "routingKey";
+
+ @Mock ConnectionFactory defaultConnectionFactory;
+ @Mock ConnectionFactory routedConnectionFactory;
+
+ DummyRoutingConnectionFactory sut;
+
+ @Before
+ public void before() {
+
+ sut = new DummyRoutingConnectionFactory();
+ sut.setDefaultTargetConnectionFactory(defaultConnectionFactory);
+ }
+
+ @Test // gh-98
+ public void shouldDetermineRoutedFactory() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", routedConnectionFactory));
+ sut.setConnectionFactoryLookup(new MapConnectionFactoryLookup());
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "key")) //
+ .as(StepVerifier::create) //
+ .expectNext(routedConnectionFactory) //
+ .verifyComplete();
+ }
+
+ @Test // gh-98
+ public void shouldFallbackToDefaultConnectionFactory() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", routedConnectionFactory));
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .as(StepVerifier::create) //
+ .expectNext(defaultConnectionFactory) //
+ .verifyComplete();
+ }
+
+ @Test // gh-98
+ public void initializationShouldFailUnsupportedLookupKey() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", new Object()));
+
+ assertThatThrownBy(() -> sut.afterPropertiesSet()).isInstanceOf(IllegalArgumentException.class);
+ }
+
+ @Test // gh-98
+ public void initializationShouldFailUnresolvableKey() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", "value"));
+ sut.setConnectionFactoryLookup(new MapConnectionFactoryLookup());
+
+ assertThatThrownBy(() -> sut.afterPropertiesSet()).isInstanceOf(ConnectionFactoryLookupFailureException.class)
+ .hasMessageContaining("No ConnectionFactory with name 'value' registered");
+ }
+
+ @Test // gh-98
+ public void unresolvableConnectionFactoryRetrievalShouldFail() {
+
+ sut.setLenientFallback(false);
+ sut.setConnectionFactoryLookup(new MapConnectionFactoryLookup());
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", routedConnectionFactory));
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "unknown")) //
+ .as(StepVerifier::create) //
+ .verifyError(IllegalStateException.class);
+ }
+
+ @Test // gh-98
+ public void connectionFactoryRetrievalWithUnknownLookupKeyShouldReturnDefaultConnectionFactory() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", routedConnectionFactory));
+ sut.setDefaultTargetConnectionFactory(defaultConnectionFactory);
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "unknown")) //
+ .as(StepVerifier::create) //
+ .expectNext(defaultConnectionFactory) //
+ .verifyComplete();
+ }
+
+ @Test // gh-98
+ public void connectionFactoryRetrievalWithoutLookupKeyShouldReturnDefaultConnectionFactory() {
+
+ sut.setTargetConnectionFactories(Collections.singletonMap("key", routedConnectionFactory));
+ sut.setDefaultTargetConnectionFactory(defaultConnectionFactory);
+ sut.setLenientFallback(false);
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .as(StepVerifier::create) //
+ .expectNext(defaultConnectionFactory) //
+ .verifyComplete();
+ }
+
+ @Test // gh-98
+ public void shouldLookupFromMap() {
+
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup("lookup-key", routedConnectionFactory);
+
+ sut.setConnectionFactoryLookup(lookup);
+ sut.setTargetConnectionFactories(Collections.singletonMap("my-key", "lookup-key"));
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "my-key")) //
+ .as(StepVerifier::create) //
+ .expectNext(routedConnectionFactory) //
+ .verifyComplete();
+ }
+
+ @Test // gh-98
+ @SuppressWarnings("unchecked")
+ public void shouldAllowModificationsAfterInitialization() {
+
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup();
+
+ sut.setConnectionFactoryLookup(lookup);
+ sut.setTargetConnectionFactories((Map) lookup.getConnectionFactories());
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "lookup-key")) //
+ .as(StepVerifier::create) //
+ .expectNext(defaultConnectionFactory) //
+ .verifyComplete();
+
+ lookup.addConnectionFactory("lookup-key", routedConnectionFactory);
+ sut.afterPropertiesSet();
+
+ sut.determineTargetConnectionFactory() //
+ .subscriberContext(Context.of(ROUTING_KEY, "lookup-key")) //
+ .as(StepVerifier::create) //
+ .expectNext(routedConnectionFactory) //
+ .verifyComplete();
+ }
+
+ static class DummyRoutingConnectionFactory extends AbstractRoutingConnectionFactory {
+
+ @Override
+ protected Mono determineCurrentLookupKey() {
+ return Mono.subscriberContext().filter(it -> it.hasKey(ROUTING_KEY)).map(it -> it.get(ROUTING_KEY));
+ }
+ }
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookupUnitTests.java b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookupUnitTests.java
new file mode 100644
index 00000000..10b212fd
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookupUnitTests.java
@@ -0,0 +1,80 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
+
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanNotOfRequiredTypeException;
+
+/**
+ * Unit tests for {@link BeanFactoryConnectionFactoryLookup}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class BeanFactoryConnectionFactoryLookupUnitTests {
+
+ private static final String CONNECTION_FACTORY_BEAN_NAME = "connectionFactory";
+
+ @Mock BeanFactory beanFactory;
+
+ @Test // gh-98
+ public void shouldLookupConnectionFactory() {
+
+ DummyConnectionFactory expectedConnectionFactory = new DummyConnectionFactory();
+ when(beanFactory.getBean(CONNECTION_FACTORY_BEAN_NAME, ConnectionFactory.class))
+ .thenReturn(expectedConnectionFactory);
+
+ BeanFactoryConnectionFactoryLookup lookup = new BeanFactoryConnectionFactoryLookup();
+ lookup.setBeanFactory(beanFactory);
+ ConnectionFactory connectionFactory = lookup.getConnectionFactory(CONNECTION_FACTORY_BEAN_NAME);
+
+ assertThat(connectionFactory).isNotNull();
+ assertThat(connectionFactory).isSameAs(expectedConnectionFactory);
+ }
+
+ @Test // gh-98
+ public void shouldLookupWhereBeanFactoryYieldsNonConnectionFactoryType() {
+
+ BeanFactory beanFactory = mock(BeanFactory.class);
+
+ when(beanFactory.getBean(CONNECTION_FACTORY_BEAN_NAME, ConnectionFactory.class)).thenThrow(
+ new BeanNotOfRequiredTypeException(CONNECTION_FACTORY_BEAN_NAME, ConnectionFactory.class, String.class));
+
+ BeanFactoryConnectionFactoryLookup lookup = new BeanFactoryConnectionFactoryLookup(beanFactory);
+
+ assertThatExceptionOfType(ConnectionFactoryLookupFailureException.class)
+ .isThrownBy(() -> lookup.getConnectionFactory(CONNECTION_FACTORY_BEAN_NAME));
+ }
+
+ @Test // gh-98
+ public void shouldLookupWhereBeanFactoryHasNotBeenSupplied() {
+
+ BeanFactoryConnectionFactoryLookup lookup = new BeanFactoryConnectionFactoryLookup();
+
+ assertThatThrownBy(() -> lookup.getConnectionFactory(CONNECTION_FACTORY_BEAN_NAME))
+ .isInstanceOf(IllegalStateException.class);
+ }
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/DummyConnectionFactory.java b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/DummyConnectionFactory.java
new file mode 100644
index 00000000..a7ba01cd
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/DummyConnectionFactory.java
@@ -0,0 +1,42 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import io.r2dbc.spi.Connection;
+import io.r2dbc.spi.ConnectionFactory;
+import io.r2dbc.spi.ConnectionFactoryMetadata;
+
+import org.reactivestreams.Publisher;
+
+/**
+ * Stub, do-nothing {@link ConnectionFactory} implementation.
+ *
+ * All methods throw {@link UnsupportedOperationException}.
+ *
+ * @author Mark Paluch
+ */
+class DummyConnectionFactory implements ConnectionFactory {
+
+ @Override
+ public Publisher create() {
+ throw new UnsupportedOperationException();
+ }
+
+ @Override
+ public ConnectionFactoryMetadata getMetadata() {
+ throw new UnsupportedOperationException();
+ }
+}
diff --git a/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookupUnitTests.java b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookupUnitTests.java
new file mode 100644
index 00000000..a4357684
--- /dev/null
+++ b/src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookupUnitTests.java
@@ -0,0 +1,102 @@
+/*
+ * 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.connectionfactory.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+
+import io.r2dbc.spi.ConnectionFactory;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.junit.Test;
+
+/**
+ * Unit tests for {@link MapConnectionFactoryLookup}.
+ *
+ * @author Mark Paluch
+ */
+public class MapConnectionFactoryLookupUnitTests {
+
+ private static final String CONNECTION_FACTORY_NAME = "connectionFactory";
+
+ @Test // gh-98
+ public void getConnectionFactorysReturnsUnmodifiableMap() {
+
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup();
+ Map connectionFactories = lookup.getConnectionFactories();
+
+ assertThatThrownBy(() -> connectionFactories.put("", new DummyConnectionFactory()))
+ .isInstanceOf(UnsupportedOperationException.class);
+ }
+
+ @Test // gh-98
+ public void shouldLookupConnectionFactory() {
+
+ Map connectionFactories = new HashMap<>();
+ DummyConnectionFactory expectedConnectionFactory = new DummyConnectionFactory();
+
+ connectionFactories.put(CONNECTION_FACTORY_NAME, expectedConnectionFactory);
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup();
+
+ lookup.setConnectionFactories(connectionFactories);
+
+ ConnectionFactory connectionFactory = lookup.getConnectionFactory(CONNECTION_FACTORY_NAME);
+
+ assertThat(connectionFactory).isNotNull();
+ assertThat(connectionFactory).isSameAs(expectedConnectionFactory);
+ }
+
+ @Test // gh-98
+ public void addingConnectionFactoryPermitsOverride() {
+
+ Map connectionFactories = new HashMap<>();
+ DummyConnectionFactory overriddenConnectionFactory = new DummyConnectionFactory();
+ DummyConnectionFactory expectedConnectionFactory = new DummyConnectionFactory();
+ connectionFactories.put(CONNECTION_FACTORY_NAME, overriddenConnectionFactory);
+
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup();
+
+ lookup.setConnectionFactories(connectionFactories);
+ lookup.addConnectionFactory(CONNECTION_FACTORY_NAME, expectedConnectionFactory);
+
+ ConnectionFactory connectionFactory = lookup.getConnectionFactory(CONNECTION_FACTORY_NAME);
+
+ assertThat(connectionFactory).isNotNull();
+ assertThat(connectionFactory).isSameAs(expectedConnectionFactory);
+ }
+
+ @Test // gh-98
+ @SuppressWarnings("unchecked")
+ public void getConnectionFactoryWhereSuppliedMapHasNonConnectionFactoryTypeUnderSpecifiedKey() {
+
+ Map connectionFactories = new HashMap<>();
+ connectionFactories.put(CONNECTION_FACTORY_NAME, new Object());
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup(connectionFactories);
+
+ assertThatThrownBy(() -> lookup.getConnectionFactory(CONNECTION_FACTORY_NAME))
+ .isInstanceOf(ClassCastException.class);
+ }
+
+ @Test // gh-98
+ public void getConnectionFactoryWhereSuppliedMapHasNoEntryForSpecifiedKey() {
+
+ MapConnectionFactoryLookup lookup = new MapConnectionFactoryLookup();
+
+ assertThatThrownBy(() -> lookup.getConnectionFactory(CONNECTION_FACTORY_NAME))
+ .isInstanceOf(ConnectionFactoryLookupFailureException.class);
+ }
+}