From e3aae619f416645314c9f9bd0ce5ddfa874a30b9 Mon Sep 17 00:00:00 2001 From: Mark Paluch Date: Tue, 28 May 2019 15:14:50 +0200 Subject: [PATCH] #98 - Add support for AbstractRoutingConnectionFactory. MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit We now provide an abstract base class for ConnectionFactory routing. Routing keys are typically obtained from a subscriber context. AbstractRoutingConnectionFactory is backed by either a map of string identifiers or connection factories. When using string identifiers, these can map agains e.g. Spring bean names that can be resolved using BeanFactoryConnectionFactoryLookup. class MyRoutingConnectionFactory extends AbstractRoutingConnectionFactory { @Override protected Mono determineCurrentLookupKey() { return Mono.subscriberContext().filter(it -> it.hasKey(ROUTING_KEY)).map(it -> it.get(ROUTING_KEY)); } } @Bean public void routingConnectionFactory() { MyRoutingConnectionFactory router = new MyRoutingConnectionFactory(); Map factories = new HashMap<>(); ConnectionFactory myDefault = …; ConnectionFactory primary = …; ConnectionFactory secondary = …; factories.put("primary", primary); factories.put("secondary", secondary); router.setTargetConnectionFactories(factories); router.setDefaultTargetConnectionFactory(myDefault); return router; } Original pull request: #132. --- .../AbstractRoutingConnectionFactory.java | 244 ++++++++++++++++++ .../BeanFactoryConnectionFactoryLookup.java | 90 +++++++ .../lookup/ConnectionFactoryLookup.java | 36 +++ ...nnectionFactoryLookupFailureException.java | 47 ++++ .../lookup/MapConnectionFactoryLookup.java | 123 +++++++++ .../lookup/SingleConnectionFactoryLookup.java | 53 ++++ .../lookup/package-info.java | 6 + ...ractRoutingConnectionFactoryUnitTests.java | 191 ++++++++++++++ ...ctoryConnectionFactoryLookupUnitTests.java | 80 ++++++ .../lookup/DummyConnectionFactory.java | 42 +++ .../MapConnectionFactoryLookupUnitTests.java | 102 ++++++++ 11 files changed, 1014 insertions(+) create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactory.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookup.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookup.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/ConnectionFactoryLookupFailureException.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookup.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/SingleConnectionFactoryLookup.java create mode 100644 src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/package-info.java create mode 100644 src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactoryUnitTests.java create mode 100644 src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/BeanFactoryConnectionFactoryLookupUnitTests.java create mode 100644 src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/DummyConnectionFactory.java create mode 100644 src/test/java/org/springframework/data/r2dbc/connectionfactory/lookup/MapConnectionFactoryLookupUnitTests.java diff --git a/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactory.java b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactory.java new file mode 100644 index 00000000..6a128bb0 --- /dev/null +++ b/src/main/java/org/springframework/data/r2dbc/connectionfactory/lookup/AbstractRoutingConnectionFactory.java @@ -0,0 +1,244 @@ +/* + * 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 reactor.core.publisher.Mono; + +import java.util.HashMap; +import java.util.Map; + +import org.springframework.beans.factory.InitializingBean; +import org.springframework.lang.Nullable; +import org.springframework.util.Assert; + +/** + * Abstract {@link ConnectionFactory} implementation that routes {@link #create()} calls to one of various target + * {@link ConnectionFactory factories} based on a lookup key. The latter is typically (but not necessarily) determined + * from some subscriber context. + *

+ * Allows to configure a {@link #setDefaultTargetConnectionFactory(Object) default ConnectionFactory} as fallback. + *

+ * Calls to {@link #getMetadata()} are routed to the {@link #setDefaultTargetConnectionFactory(Object) default + * ConnectionFactory} if configured. + * + * @author Mark Paluch + * @see #setTargetConnectionFactories + * @see #setDefaultTargetConnectionFactory + * @see #determineCurrentLookupKey() + */ +public abstract class AbstractRoutingConnectionFactory implements ConnectionFactory, InitializingBean { + + private static final Object FALLBACK_MARKER = new Object(); + + private @Nullable Map targetConnectionFactories; + + private @Nullable Object defaultTargetConnectionFactory; + + private boolean lenientFallback = true; + + private ConnectionFactoryLookup connectionFactoryLookup = new MapConnectionFactoryLookup(); + + private @Nullable Map 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); + } +}