DATACASS-330 - Add Session routing.

We now support Session routing with AbstractRoutingSessionFactory. Session routing is based on a map, keyed by a lookup key that is supplied by an implementing class upon Session lookup.

Extend JavaConfig to configure a SessionFactory bean and configure CqlTemplate and CassandraAdminTemplate accordingly.

spring-cql/src/main/java/org/springframework/cassandra/config/java/AbstractCqlTemplateConfiguration.java

@@ -33,13 +33,13 @@ import org.springframework.context.annotation.Bean;
 public abstract class AbstractCqlTemplateConfiguration extends AbstractSessionConfiguration {
 
 	/**
-	 * Creates a {@link CqlTemplate} configured with {@link #session()}.
-	 * 
+	 * Creates a {@link CqlTemplate} configured with {@link #sessionFactory()}.
+	 *
 	 * @return the {@link CqlTemplate}.
-	 * @see #session()
+	 * @see #sessionFactory()
 	 */
 	@Bean
 	public CqlTemplate cqlTemplate() {
-		return new CqlTemplate(session().getObject());
+		return new CqlTemplate(sessionFactory());
 	}
 }

spring-cql/src/main/java/org/springframework/cassandra/config/java/AbstractSessionConfiguration.java

@@ -16,6 +16,8 @@
 package org.springframework.cassandra.config.java;
 
 import org.springframework.cassandra.config.CassandraCqlSessionFactoryBean;
+import org.springframework.cassandra.core.session.DefaultSessionFactory;
+import org.springframework.cassandra.core.session.SessionFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 
@@ -52,6 +54,18 @@ public abstract class AbstractSessionConfiguration extends AbstractClusterConfig
 		return bean;
 	}
 
+	/**
+	 * Creates a {@link DefaultSessionFactory} using the configured {@link #session()} to be used with
+	 * {@link org.springframework.cassandra.core.CqlTemplate}.
+	 *
+	 * @return {@link SessionFactory} used to initialize the Template API.
+	 * @since 2.0
+	 */
+	@Bean
+	public SessionFactory sessionFactory() {
+		return new DefaultSessionFactory(session().getObject());
+	}
+
 	/**
 	 * Return the name of the keyspace to connect to.
 	 *

spring-cql/src/main/java/org/springframework/cassandra/core/session/SessionFactory.java

@@ -37,7 +37,7 @@ public interface SessionFactory {
 	/**
 	 * Attempts to establish a {@link Session} with the connection infrastructure that this {@link SessionFactory} object
 	 * represents.
-	 * 
+	 *
 	 * @return a {@link Session} to Apache Cassandra.
 	 */
 	Session getSession();

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/AbstractRoutingSessionFactory.java

@@ -0,0 +1,216 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.beans.factory.InitializingBean;
+import org.springframework.cassandra.core.session.SessionFactory;
+import org.springframework.util.Assert;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Abstract {@link org.springframework.cassandra.core.session.SessionFactory} implementation that routes
+ * {@link #getSession()} calls to one of various target {@link SessionFactory factories} based on a lookup key. The
+ * latter is usually (but not necessarily) determined through some thread-bound transaction context.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see #setTargetSessionFactories(Map)
+ * @see #setDefaultTargetSessionFactory(Object)
+ * @see #determineCurrentLookupKey()
+ * @see SessionFactoryLookup
+ */
+public abstract class AbstractRoutingSessionFactory implements SessionFactory, InitializingBean {
+
+	private Map<Object, Object> targetSessionFactories;
+
+	private Object defaultTargetSessionFactory;
+
+	private boolean lenientFallback = true;
+
+	private SessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup();
+
+	private Map<Object, SessionFactory> resolvedSessionFactories;
+
+	private SessionFactory resolvedDefaultSessionFactory;
+
+	/**
+	 * Specify the map of target session factories, with the lookup key as key.
+	 * <p>
+	 * The mapped value can either be a corresponding {@link SessionFactory} instance or a data source name String (to be
+	 * resolved via a {@link #setSessionFactoryLookup(SessionFactoryLookup)}).
+	 * <p>
+	 * 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()}.
+	 */
+	public void setTargetSessionFactories(Map<Object, Object> targetSessionFactories) {
+		this.targetSessionFactories = targetSessionFactories;
+	}
+
+	/**
+	 * Specify the default target {@link SessionFactory}, if any.
+	 * <p>
+	 * The mapped value can either be a corresponding {@link SessionFactory} instance or a data source name String (to be
+	 * resolved via a {@link #setSessionFactoryLookup(SessionFactoryLookup)}).
+	 * <p>
+	 * This {@link SessionFactory} will be used as target if none of the keyed {@link #setTargetSessionFactories(Map)}
+	 * match the {@link #determineCurrentLookupKey()} current lookup key.
+	 */
+	public void setDefaultTargetSessionFactory(Object defaultTargetSessionFactory) {
+		this.defaultTargetSessionFactory = defaultTargetSessionFactory;
+	}
+
+	/**
+	 * Specify whether to apply a lenient fallback to the default {@link SessionFactory} if no specific
+	 * {@link SessionFactory} could be found for the current lookup key.
+	 * <p>
+	 * Default is {@literal true}, accepting lookup keys without a corresponding entry in the target
+	 * {@link SessionFactory} map - simply falling back to the default {@link SessionFactory} in that case.
+	 * <p>
+	 * Switch this flag to {@literal false} if you would prefer the fallback to only apply if the lookup key was
+	 * {@code null}. Lookup keys without a {@link SessionFactory} entry will then lead to an
+	 * {@link IllegalStateException}.
+	 *
+	 * @param lenientFallback {@literal true} to accepting lookup keys without a corresponding entry in the target.
+	 * @see #setTargetSessionFactories
+	 * @see #setDefaultTargetSessionFactory
+	 * @see #determineCurrentLookupKey()
+	 */
+	public void setLenientFallback(boolean lenientFallback) {
+		this.lenientFallback = lenientFallback;
+	}
+
+	/**
+	 * Set the {@link SessionFactoryLookup} implementation to use for resolving session factory name Strings in the
+	 * {@link #setTargetSessionFactories(Map)} map.
+	 * <p>
+	 * Default is a {@link MapSessionFactoryLookup}, allowing a string keyed map of {@link SessionFactory session
+	 * factories}.
+	 *
+	 * @param sessionFactoryLookup the {@link SessionFactoryLookup}. Defaults to {@link MapSessionFactoryLookup} if
+	 *          {@literal null}.
+	 */
+	public void setSessionFactoryLookup(SessionFactoryLookup sessionFactoryLookup) {
+		this.sessionFactoryLookup = (sessionFactoryLookup != null ? sessionFactoryLookup : new MapSessionFactoryLookup());
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.cassandra.core.session.SessionFactory#getSession()
+	 */
+	@Override
+	public Session getSession() {
+		return determineTargetSessionFactory().getSession();
+	}
+
+	// -------------------------------------------------------------------------
+	// Implementation hooks and helper methods
+	// -------------------------------------------------------------------------
+
+	@Override
+	public void afterPropertiesSet() {
+
+		Assert.notNull(this.targetSessionFactories, "Property targetSessionFactories is required");
+
+		this.resolvedSessionFactories = new HashMap<>(this.targetSessionFactories.size());
+		for (Map.Entry<Object, Object> entry : this.targetSessionFactories.entrySet()) {
+
+			Object lookupKey = resolveSpecifiedLookupKey(entry.getKey());
+			SessionFactory sessionFactory = resolveSpecifiedSessionFactory(entry.getValue());
+			this.resolvedSessionFactories.put(lookupKey, sessionFactory);
+		}
+
+		if (this.defaultTargetSessionFactory != null) {
+			this.resolvedDefaultSessionFactory = resolveSpecifiedSessionFactory(this.defaultTargetSessionFactory);
+		}
+	}
+
+	/**
+	 * Resolve the given lookup key object, as specified in the {@link #setTargetSessionFactories(Map)} map, into the
+	 * actual lookup key to be used for matching with the {@link #determineCurrentLookupKey() current lookup key}.
+	 * <p>
+	 * 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 {@code sessionFactory} object into a {@link SessionFactory} instance.
+	 * <p>
+	 * The default implementation handles {@link SessionFactory} instances and session factory names (to be resolved via a
+	 * {@link #setSessionFactoryLookup(SessionFactoryLookup)}).
+	 *
+	 * @param sessionFactory the session factory value object as specified in the {@link #setTargetSessionFactories(Map)}
+	 *          map
+	 * @return the resolved {@link SessionFactory}
+	 * @throws IllegalArgumentException in case of an unsupported value type.
+	 * @throws SessionFactoryLookupFailureException if the lookup failed.
+	 */
+	protected SessionFactory resolveSpecifiedSessionFactory(Object sessionFactory) throws IllegalArgumentException {
+
+		if (sessionFactory instanceof SessionFactory) {
+			return (SessionFactory) sessionFactory;
+		} else if (sessionFactory instanceof String) {
+			return this.sessionFactoryLookup.getSessionFactory((String) sessionFactory);
+		} else {
+			throw new IllegalArgumentException(String
+					.format("Illegal session factory value. Only [org.springframework.cassandra.core.session.SessionFactory] "
+							+ "and String supported: %s", sessionFactory));
+		}
+	}
+
+	/**
+	 * Retrieve the current target {@link SessionFactory}. Determines the {@link #determineCurrentLookupKey() current
+	 * lookup key}, performs a lookup in the {@link #setTargetSessionFactories(Map)} map, falls back to the specified
+	 * {@link #setDefaultTargetSessionFactory default target SessionFactory} if necessary.
+	 *
+	 * @see #determineCurrentLookupKey()
+	 */
+	protected SessionFactory determineTargetSessionFactory() {
+
+		Assert.notNull(this.resolvedSessionFactories, "SessionFactory router not initialized");
+
+		Object lookupKey = determineCurrentLookupKey();
+		SessionFactory sessionFactory = this.resolvedSessionFactories.get(lookupKey);
+
+		if (sessionFactory == null && (this.lenientFallback || lookupKey == null)) {
+			sessionFactory = this.resolvedDefaultSessionFactory;
+		}
+
+		if (sessionFactory == null) {
+			throw new IllegalStateException(
+					String.format("Cannot determine target SessionFactory for lookup key [%s]", lookupKey));
+		}
+
+		return sessionFactory;
+	}
+
+	/**
+	 * Determine the current lookup key. This will typically be implemented to check a thread-bound context.
+	 * <p>
+	 * Allows for arbitrary keys.
+	 *
+	 * @return the current lookup key. The returned key needs to match the stored lookup key type.
+	 */
+	protected abstract Object determineCurrentLookupKey();
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/BeanFactorySessionFactoryLookup.java

@@ -0,0 +1,88 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import org.springframework.beans.BeansException;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.BeanFactoryAware;
+import org.springframework.cassandra.core.session.SessionFactory;
+import org.springframework.util.Assert;
+
+/**
+ * {@link SessionFactoryLookup} implementation based on a Spring {@link BeanFactory}.
+ * <p>
+ * Will lookup Spring managed beans identified by bean name, expecting them to be of type
+ * {@link org.springframework.cassandra.core.session.SessionFactory}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see org.springframework.beans.factory.BeanFactory
+ */
+public class BeanFactorySessionFactoryLookup implements SessionFactoryLookup, BeanFactoryAware {
+
+	private BeanFactory beanFactory;
+
+	/**
+	 * Create a new instance of {@link BeanFactorySessionFactoryLookup}.
+	 * <p>
+	 * The BeanFactory to access must be set via {@link #setBeanFactory(BeanFactory)}.
+	 *
+	 * @see #setBeanFactory(BeanFactory)
+	 */
+	public BeanFactorySessionFactoryLookup() {}
+
+	/**
+	 * Create a new instance of {@link BeanFactorySessionFactoryLookup} given {@link BeanFactory}.
+	 * <p>
+	 * 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 ({@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 org.springframework.cassandra.core.session.SessionFactory session factories}, must not be
+	 *          {@literal null}.
+	 */
+	public BeanFactorySessionFactoryLookup(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.cassandra.core.session.lookup.SessionFactoryLookup#getSessionFactory(java.lang.String)
+	 */
+	@Override
+	public SessionFactory getSessionFactory(String sessionFactoryName) throws SessionFactoryLookupFailureException {
+
+		Assert.notNull(this.beanFactory, "BeanFactory must not be null");
+
+		try {
+			return this.beanFactory.getBean(sessionFactoryName, SessionFactory.class);
+		} catch (BeansException ex) {
+			throw new SessionFactoryLookupFailureException(
+					String.format("Failed to look up SessionFactory bean with name [%s]", sessionFactoryName), ex);
+		}
+	}
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/MapSessionFactoryLookup.java

@@ -0,0 +1,120 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+import org.springframework.cassandra.core.session.SessionFactory;
+import org.springframework.util.Assert;
+
+/**
+ * Simple {@link SessionFactoryLookup} implementation that relies on a map for doing lookups.
+ * <p>
+ * Useful for testing environments or applications that need to match arbitrary {@link String} names to target
+ * {@link SessionFactory} objects. This class is not thread-safe for modifications. Once initialized, it can be shared
+ * amongst multiple threads and is thread-safe for reading.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+public class MapSessionFactoryLookup implements SessionFactoryLookup {
+
+	private final Map<String, SessionFactory> sessionFactories = new HashMap<>(4);
+
+	/**
+	 * Create a new instance of {@link MapSessionFactoryLookup}.
+	 */
+	public MapSessionFactoryLookup() {}
+
+	/**
+	 * Create a new instance of {@link MapSessionFactoryLookup}.
+	 *
+	 * @param sessionFactories the {@link Map} of {@link SessionFactory session factories}. The keys are {@link String
+	 *          Strings}, the values are actual {@link SessionFactory} instances.
+	 */
+	public MapSessionFactoryLookup(Map<String, SessionFactory> sessionFactories) {
+		setSessionFactories(sessionFactories);
+	}
+
+	/**
+	 * Create a new instance of {@link MapSessionFactoryLookup}.
+	 *
+	 * @param sessionFactoryName the name under which the supplied {@link SessionFactory} is to be added
+	 * @param sessionFactory the {@link SessionFactory} to be added
+	 */
+	public MapSessionFactoryLookup(String sessionFactoryName, SessionFactory sessionFactory) {
+		addSessionFactory(sessionFactoryName, sessionFactory);
+	}
+
+	/**
+	 * Set the {@link Map} of {@link SessionFactory session factories}; the keys are {@link String Strings}, the values
+	 * are actual {@link SessionFactory} instances.
+	 * <p>
+	 * If the supplied {@link Map} is {@code null}, then this method call effectively has no effect.
+	 *
+	 * @param sessionFactories {@link Map} of {@link SessionFactory session factories}.
+	 */
+	public void setSessionFactories(Map<String, SessionFactory> sessionFactories) {
+		if (sessionFactories != null) {
+			this.sessionFactories.putAll(sessionFactories);
+		}
+	}
+
+	/**
+	 * Get the {@link Map} of {@link SessionFactory session factories} maintained by this object.
+	 * <p>
+	 * The returned {@link Map} is {@link Collections#unmodifiableMap(java.util.Map) unmodifiable}.
+	 *
+	 * @return {@link Map} of {@link SessionFactory session factories}.
+	 */
+	public Map<String, SessionFactory> getSessionFactories() {
+		return Collections.unmodifiableMap(this.sessionFactories);
+	}
+
+	/**
+	 * Add the supplied {@link SessionFactory} to the map of {@link SessionFactory session factories} maintained by this
+	 * object.
+	 *
+	 * @param sessionFactoryName the name under which the supplied {@link SessionFactory} is to be added
+	 * @param sessionFactory the {@link SessionFactory} to be so added
+	 */
+	public void addSessionFactory(String sessionFactoryName, SessionFactory sessionFactory) {
+
+		Assert.notNull(sessionFactoryName, "SessionFactory name must not be null");
+		Assert.notNull(sessionFactory, "SessionFactory must not be null");
+
+		this.sessionFactories.put(sessionFactoryName, sessionFactory);
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.cassandra.core.session.lookup.SessionFactoryLookup#getSessionFactory(java.lang.String)
+	 */
+	@Override
+	public SessionFactory getSessionFactory(String sessionFactoryName) throws SessionFactoryLookupFailureException {
+
+		Assert.notNull(sessionFactoryName, "SessionFactory name must not be null");
+
+		SessionFactory sessionFactory = this.sessionFactories.get(sessionFactoryName);
+		if (sessionFactory == null) {
+			throw new SessionFactoryLookupFailureException(
+					String.format("No SessionFactory with name [%s] registered", sessionFactoryName));
+		}
+
+		return sessionFactory;
+	}
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/SessionFactoryLookup.java

@@ -0,0 +1,42 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import org.springframework.cassandra.core.session.SessionFactory;
+
+/**
+ * Strategy interface for looking up {@link org.springframework.cassandra.core.session.SessionFactory} by name.
+ * <p>
+ * Implementing classes resolve session factories keyed by {@link String} from an underlying source such as a
+ * {@link java.util.Map} or the {@link org.springframework.beans.factory.BeanFactory}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ * @see AbstractRoutingSessionFactory
+ */
+@FunctionalInterface
+public interface SessionFactoryLookup {
+
+	/**
+	 * Implementations must implement this method to retrieve the {@link SessionFactory} identified by the given name from
+	 * their backing store.
+	 *
+	 * @param sessionFactoryName the name of the {@link SessionFactory}.
+	 * @return the {@link SessionFactory} (never {@code null})
+	 * @throws SessionFactoryLookupFailureException if the lookup failed.
+	 */
+	SessionFactory getSessionFactory(String sessionFactoryName) throws SessionFactoryLookupFailureException;
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/SessionFactoryLookupFailureException.java

@@ -0,0 +1,48 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import org.springframework.dao.NonTransientDataAccessException;
+
+/**
+ * Exception to be thrown by a {@link SessionFactoryLookup} implementation, indicating that the specified
+ * {@link org.springframework.cassandra.core.session.SessionFactory} could not be obtained.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+@SuppressWarnings("serial")
+public class SessionFactoryLookupFailureException extends NonTransientDataAccessException {
+
+	/**
+	 * Create a new {@link SessionFactoryLookupFailureException}.
+	 *
+	 * @param msg the detail message.
+	 */
+	public SessionFactoryLookupFailureException(String msg) {
+		super(msg);
+	}
+
+	/**
+	 * Create a new {@link SessionFactoryLookupFailureException}.
+	 *
+	 * @param msg the detail message.
+	 * @param cause the root cause (usually from using a underlying lookup API).
+	 */
+	public SessionFactoryLookupFailureException(String msg, Throwable cause) {
+		super(msg, cause);
+	}
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/SingleSessionFactoryLookup.java

@@ -0,0 +1,51 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import org.springframework.cassandra.core.session.SessionFactory;
+import org.springframework.util.Assert;
+
+/**
+ * An implementation of {@link SessionFactoryLookup} that simply wraps a single given {@link SessionFactory}, returned
+ * for any session factory name. Useful for testing or environments that provide only one {@link SessionFactory}.
+ *
+ * @author Mark Paluch
+ * @since 2.0
+ */
+public class SingleSessionFactoryLookup implements SessionFactoryLookup {
+
+	private final SessionFactory sessionFactory;
+
+	/**
+	 * Create a new instance of {@link SingleSessionFactoryLookup} given {@link SessionFactory}.
+	 *
+	 * @param sessionFactory the single {@link SessionFactory} to wrap, must not be {@literal null}.
+	 */
+	public SingleSessionFactoryLookup(SessionFactory sessionFactory) {
+
+		Assert.notNull(sessionFactory, "SessionFactory must not be null");
+
+		this.sessionFactory = sessionFactory;
+	}
+
+	/* (non-Javadoc)
+	 * @see org.springframework.cassandra.core.session.lookup.SessionFactoryLookup#getSessionFactory(java.lang.String)
+	 */
+	@Override
+	public SessionFactory getSessionFactory(String sessionFactoryName) throws SessionFactoryLookupFailureException {
+		return sessionFactory;
+	}
+}

spring-cql/src/main/java/org/springframework/cassandra/core/session/lookup/package-info.java

@@ -0,0 +1,6 @@
+/**
+ * Provides a strategy for looking up {@link org.springframework.cassandra.core.session.SessionFactory}.
+ * 
+ * @author Mark Paluch
+ */
+package org.springframework.cassandra.core.session.lookup;

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/AbstractRoutingSessionFactoryUnitTests.java

@@ -0,0 +1,174 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+
+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.runners.MockitoJUnitRunner;
+import org.springframework.cassandra.core.session.DefaultSessionFactory;
+
+import com.datastax.driver.core.Session;
+
+/**
+ * Unit tests for {@link AbstractRoutingSessionFactory}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class AbstractRoutingSessionFactoryUnitTests {
+
+	@Mock Session defaultSession;
+	@Mock Session routedSession;
+
+	StubbedRoutingSessionFactory sut;
+
+	@Before
+	public void before() throws Exception {
+
+		sut = new StubbedRoutingSessionFactory();
+
+		sut.setDefaultTargetSessionFactory(new DefaultSessionFactory(defaultSession));
+	}
+
+	@Test // DATACASS-330
+	public void shouldDetermineRoutedRepository() {
+
+		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey("key");
+
+		assertThat(sut.getSession()).isSameAs(routedSession);
+	}
+
+	@Test // DATACASS-330
+	public void shouldFallbackToDefaultSession() {
+
+		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey("unknown");
+
+		assertThat(sut.getSession()).isSameAs(defaultSession);
+	}
+
+	@Test // DATACASS-330
+	public void initializationShouldFailUnsupportedLookupKey() {
+
+		sut.setTargetSessionFactories(Collections.singletonMap("key", new Object()));
+		try {
+			sut.afterPropertiesSet();
+			fail("Missing IllegalArgumentException");
+		} catch (IllegalArgumentException e) {
+			assertThat(e).hasMessageContaining("Illegal session factory value.");
+		}
+	}
+
+	@Test // DATACASS-330
+	public void initializationShouldFailUnresolvableKey() {
+
+		sut.setTargetSessionFactories(Collections.singletonMap("key", "value"));
+		sut.setSessionFactoryLookup(new MapSessionFactoryLookup());
+
+		try {
+			sut.afterPropertiesSet();
+			fail("Missing SessionFactoryLookupFailureException");
+		} catch (SessionFactoryLookupFailureException e) {
+			assertThat(e).hasMessageContaining("No SessionFactory with name [value] registered");
+		}
+	}
+
+	@Test // DATACASS-330
+	public void unresolvableSessionRetrievalShouldFail() {
+
+		sut.setLenientFallback(false);
+		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey("unknown");
+
+		try {
+			sut.getSession();
+			fail("Missing IllegalStateException");
+		} catch (RuntimeException e) {}
+	}
+
+	@Test // DATACASS-330
+	public void sessionRetrievalWithoutLookupKeyShouldReturnDefaultSession() {
+
+		sut.setTargetSessionFactories(Collections.singletonMap("key", new DefaultSessionFactory(routedSession)));
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey(null);
+
+		assertThat(sut.getSession()).isSameAs(defaultSession);
+	}
+
+	@Test // DATACASS-330
+	public void shouldLookupFromMap() {
+
+		MapSessionFactoryLookup lookup = new MapSessionFactoryLookup("lookup-key",
+				new DefaultSessionFactory(routedSession));
+
+		sut.setSessionFactoryLookup(lookup);
+		sut.setTargetSessionFactories(Collections.singletonMap("my-key", "lookup-key"));
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey("my-key");
+
+		assertThat(sut.getSession()).isSameAs(routedSession);
+	}
+
+	@Test // DATACASS-330
+	public void shouldAllowModificationsAfterInitialization() {
+
+		MapSessionFactoryLookup lookup = new MapSessionFactoryLookup();
+
+		sut.setSessionFactoryLookup(lookup);
+		sut.setTargetSessionFactories((Map) lookup.getSessionFactories());
+		sut.afterPropertiesSet();
+
+		sut.setLookupKey("lookup-key");
+
+		assertThat(sut.getSession()).isSameAs(defaultSession);
+
+		lookup.addSessionFactory("lookup-key", new DefaultSessionFactory(routedSession));
+		sut.afterPropertiesSet();
+
+		assertThat(sut.getSession()).isSameAs(routedSession);
+	}
+
+	static class StubbedRoutingSessionFactory extends AbstractRoutingSessionFactory {
+
+		String lookupKey;
+
+		void setLookupKey(String lookupKey) {
+			this.lookupKey = lookupKey;
+		}
+
+		@Override
+		protected Object determineCurrentLookupKey() {
+			return lookupKey;
+		}
+	}
+}

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/BeanFactorySessionFactoryLookupUnitTests.java

@@ -0,0 +1,74 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.beans.factory.BeanFactory;
+import org.springframework.beans.factory.NoSuchBeanDefinitionException;
+import org.springframework.cassandra.core.session.SessionFactory;
+
+/**
+ * Unit test for {@link BeanFactorySessionFactoryLookup}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class BeanFactorySessionFactoryLookupUnitTests {
+
+	@Mock BeanFactory beanFactory;
+	@Mock SessionFactory sessionFactory;
+
+	@Test(expected = IllegalArgumentException.class) // DATACASS-330
+	public void shouldRejectNullBeanFactory() {
+		new BeanFactorySessionFactoryLookup(null);
+	}
+
+	@Test // DATACASS-330
+	public void shouldResolveSessionFactoryFromBeanFactory() throws Exception {
+
+		when(beanFactory.getBean("factory", SessionFactory.class)).thenReturn(sessionFactory);
+
+		BeanFactorySessionFactoryLookup lookup = new BeanFactorySessionFactoryLookup();
+		lookup.setBeanFactory(beanFactory);
+
+		SessionFactory result = lookup.getSessionFactory("factory");
+
+		assertThat(result).isSameAs(sessionFactory);
+	}
+
+	@Test // DATACASS-330
+	public void shouldThrowExceptionIfLookupFails() throws Exception {
+
+		when(beanFactory.getBean("factory", SessionFactory.class)).thenThrow(new NoSuchBeanDefinitionException("factory"));
+
+		BeanFactorySessionFactoryLookup lookup = new BeanFactorySessionFactoryLookup();
+		lookup.setBeanFactory(beanFactory);
+
+		try {
+			lookup.getSessionFactory("factory");
+			fail("Missing SessionFactoryLookupFailureException");
+		} catch (SessionFactoryLookupFailureException e) {
+			assertThat(e).hasMessageContaining("Failed to look up SessionFactory bean with name [factory]")
+					.hasRootCauseInstanceOf(NoSuchBeanDefinitionException.class);
+		}
+	}
+}

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/MapSessionFactoryLookupUnitTests.java

@@ -0,0 +1,85 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+
+import java.util.Collections;
+
+import org.junit.Test;
+import org.junit.runner.RunWith;
+import org.mockito.Mock;
+import org.mockito.runners.MockitoJUnitRunner;
+import org.springframework.cassandra.core.session.SessionFactory;
+
+/**
+ * Unit tests for {@link MapSessionFactoryLookup}.
+ *
+ * @author Mark Paluch
+ */
+@RunWith(MockitoJUnitRunner.class)
+public class MapSessionFactoryLookupUnitTests {
+
+	@Mock SessionFactory sessionFactory;
+
+	@Test // DATACASS-330
+	public void shouldFailWithUnknownLookup() {
+
+		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup();
+
+		try {
+			sessionFactoryLookup.getSessionFactory("unknown");
+			fail("Missing SessionFactoryLookupFailureException");
+		} catch (SessionFactoryLookupFailureException e) {
+			assertThat(e).hasMessageContaining("No SessionFactory with name [unknown] registered");
+		}
+	}
+
+	@Test // DATACASS-330
+	public void shouldResolveSessionFactoryCorrectly() {
+
+		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup("factory", sessionFactory);
+		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
+	}
+
+	@Test // DATACASS-330
+	public void shouldResolveProvidedInConstructorSessionFactoryCorrectly() {
+
+		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup(
+				Collections.singletonMap("factory", sessionFactory));
+		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
+	}
+
+	@Test // DATACASS-330
+	public void shouldSetSessionFactories() {
+
+		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup();
+
+		sessionFactoryLookup.setSessionFactories(Collections.singletonMap("factory", sessionFactory));
+
+		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
+	}
+
+	@Test // DATACASS-330
+	public void shouldNotOverwriteFactoriesSettingNull() {
+
+		MapSessionFactoryLookup sessionFactoryLookup = new MapSessionFactoryLookup("factory", sessionFactory);
+
+		sessionFactoryLookup.setSessionFactories(null);
+
+		assertThat(sessionFactoryLookup.getSessionFactory("factory")).isSameAs(sessionFactory);
+	}
+}

spring-cql/src/test/java/org/springframework/cassandra/core/session/lookup/SingleSessionFactoryLookupUnitTests.java

@@ -0,0 +1,44 @@
+/*
+ * Copyright 2017 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
+ *
+ *      http://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.cassandra.core.session.lookup;
+
+import static org.assertj.core.api.Assertions.*;
+import static org.mockito.Mockito.*;
+
+import org.junit.Test;
+import org.springframework.cassandra.core.session.SessionFactory;
+
+/**
+ * Unit tests for {@link SingleSessionFactoryLookup}.
+ *
+ * @author Mark Paluch
+ */
+public class SingleSessionFactoryLookupUnitTests {
+
+	@Test(expected = IllegalArgumentException.class) // DATACASS-330
+	public void shouldRejectNullSessionFactory() {
+		new SingleSessionFactoryLookup(null);
+	}
+
+	@Test // DATACASS-330
+	public void shouldResolveSessionFactory() {
+
+		SessionFactory sessionFactory = mock(SessionFactory.class);
+		SessionFactory result = new SingleSessionFactoryLookup(sessionFactory).getSessionFactory("any");
+
+		assertThat(result).isSameAs(sessionFactory);
+	}
+}

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractCassandraConfiguration.java

@@ -19,6 +19,8 @@ import java.util.Collections;
 
 import org.springframework.beans.factory.BeanClassLoaderAware;
 import org.springframework.cassandra.config.java.AbstractClusterConfiguration;
+import org.springframework.cassandra.core.session.DefaultSessionFactory;
+import org.springframework.cassandra.core.session.SessionFactory;
 import org.springframework.context.annotation.Bean;
 import org.springframework.context.annotation.Configuration;
 import org.springframework.core.convert.converter.Converter;
@@ -48,13 +50,13 @@ import org.springframework.data.mapping.context.MappingContext;
 public abstract class AbstractCassandraConfiguration extends AbstractClusterConfiguration
 		implements BeanClassLoaderAware {
 
-	protected ClassLoader beanClassLoader;
+	private ClassLoader beanClassLoader;
 
 	/**
 	 * Creates a {@link CassandraSessionFactoryBean} that provides a Cassandra {@link com.datastax.driver.core.Session}.
 	 * The lifecycle of {@link CassandraSessionFactoryBean} initializes the {@link #getSchemaAction() schema} in the
 	 * {@link #getKeyspaceName() configured keyspace}.
-	 * 
+	 *
 	 * @return the {@link CassandraSessionFactoryBean}.
 	 * @throws ClassNotFoundException if an error occurs initializing the initial entity set, see
 	 *           {@link #cassandraMapping()}
@@ -80,6 +82,20 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
 		return session;
 	}
 
+	/**
+	 * Creates a {@link DefaultSessionFactory} using the configured {@link #session()} to be used with
+	 * {@link org.springframework.data.cassandra.core.CassandraTemplate}.
+	 *
+	 * @return {@link SessionFactory} used to initialize the Template API.
+	 * @throws ClassNotFoundException if an error occurs initializing the initial entity set, see
+	 *           {@link #cassandraMapping()}
+	 * @since 2.0
+	 */
+	@Bean
+	public SessionFactory sessionFactory() throws ClassNotFoundException {
+		return new DefaultSessionFactory(session().getObject());
+	}
+
 	/**
 	 * Creates a {@link CassandraConverter} using the configured {@link #cassandraMapping()}. Will apply all specified
 	 * {@link #customConversions()}.
@@ -144,7 +160,7 @@ public abstract class AbstractCassandraConfiguration extends AbstractClusterConf
 	 */
 	@Bean
 	public CassandraAdminOperations cassandraTemplate() throws Exception {
-		return new CassandraAdminTemplate(session().getObject(), cassandraConverter());
+		return new CassandraAdminTemplate(sessionFactory(), cassandraConverter());
 	}
 
 	@Override

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/config/java/AbstractReactiveCassandraConfiguration.java

@@ -41,7 +41,7 @@ public abstract class AbstractReactiveCassandraConfiguration extends AbstractCas
 	 * Creates a {@link ReactiveSession} object. This wraps a {@link com.datastax.driver.core.Session} to expose Cassandra
 	 * access in a reactive style.
 	 *
-	 * @return
+	 * @return the {@link ReactiveSession}.
 	 * @see #session()
 	 * @see DefaultBridgedReactiveSession
 	 */
@@ -51,10 +51,10 @@ public abstract class AbstractReactiveCassandraConfiguration extends AbstractCas
 	}
 
 	/**
-	 * Creates a {@link ReactiveSessionFactory} to be used by the {@link ReactiveCassandraTemplate}. Will use the
+	 * Creates a {@link ReactiveSessionFactory} to be used by the {@link ReactiveCassandraTemplate}. Uses the
 	 * {@link ReactiveSession} instance configured in {@link #reactiveSession()}.
 	 *
-	 * @return
+	 * @return the {@link ReactiveSessionFactory}.
 	 * @see #reactiveSession()
 	 * @see #reactiveCassandraTemplate()
 	 */

spring-data-cassandra/src/main/java/org/springframework/data/cassandra/core/CassandraAdminTemplate.java

@@ -17,12 +17,6 @@ package org.springframework.data.cassandra.core;
 
 import java.util.Map;
 
-import com.datastax.driver.core.KeyspaceMetadata;
-import com.datastax.driver.core.Session;
-import com.datastax.driver.core.TableMetadata;
-
-import org.slf4j.Logger;
-import org.slf4j.LoggerFactory;
 import org.springframework.cassandra.core.SessionCallback;
 import org.springframework.cassandra.core.cql.CqlIdentifier;
 import org.springframework.cassandra.core.cql.generator.CreateTableCqlGenerator;
@@ -31,11 +25,16 @@ import org.springframework.cassandra.core.cql.generator.DropUserTypeCqlGenerator
 import org.springframework.cassandra.core.keyspace.CreateTableSpecification;
 import org.springframework.cassandra.core.keyspace.DropTableSpecification;
 import org.springframework.cassandra.core.keyspace.DropUserTypeSpecification;
+import org.springframework.cassandra.core.session.SessionFactory;
 import org.springframework.dao.DataAccessException;
 import org.springframework.data.cassandra.convert.CassandraConverter;
 import org.springframework.data.cassandra.mapping.CassandraPersistentEntity;
 import org.springframework.util.Assert;
 
+import com.datastax.driver.core.KeyspaceMetadata;
+import com.datastax.driver.core.Session;
+import com.datastax.driver.core.TableMetadata;
+
 /**
  * Default implementation of {@link CassandraAdminOperations}.
  *
@@ -45,10 +44,8 @@ import org.springframework.util.Assert;
  */
 public class CassandraAdminTemplate extends CassandraTemplate implements CassandraAdminOperations {
 
-	private static final Logger log = LoggerFactory.getLogger(CassandraAdminTemplate.class);
-
 	/**
-	 * Constructor used for a basic template configuration
+	 * Constructor used for a basic template configuration.
 	 *
 	 * @param session must not be {@literal null}.
 	 * @param converter must not be {@literal null}.
@@ -57,6 +54,16 @@ public class CassandraAdminTemplate extends CassandraTemplate implements Cassand
 		super(session, converter);
 	}
 
+	/**
+	 * Constructor used for a basic template configuration.
+	 *
+	 * @param sessionFactory must not be {@literal null}.
+	 * @param converter must not be {@literal null}.
+	 */
+	public CassandraAdminTemplate(SessionFactory sessionFactory, CassandraConverter converter) {
+		super(sessionFactory, converter);
+	}
+
 	/*
 	 * (non-Javadoc)
 	 * @see org.springframework.data.cassandra.core.CassandraAdminOperations#createTable(boolean, org.springframework.cassandra.core.cql.CqlIdentifier, java.lang.Class, java.util.Map)