INT-3491: (S)FTP Delegation Session Factory

JIRA: https://jira.spring.io/browse/INT-3491

Select the session factory at runtime.

Polishing; PR Comments

Also add/remove factories.

Author: garyrussell

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/CachingSessionFactory.java

@@ -1,5 +1,5 @@
 /*
- * Copyright 2002-2014 the original author or authors.
+ * Copyright 2002-2015 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.
@@ -25,6 +25,7 @@
 
 import org.springframework.beans.factory.DisposableBean;
 import org.springframework.integration.util.SimplePool;
+import org.springframework.util.Assert;
 
 /**
  * A {@link SessionFactory} implementation that caches Sessions for reuse without
@@ -62,13 +63,17 @@ public CachingSessionFactory(SessionFactory<F> sessionFactory) {
 	 * Create a CachingSessionFactory with the specified session limit. By default, if
 	 * no sessions are available in the cache, and the size limit has been reached,
 	 * calling threads will block until a session is available.
+	 * <p>
+	 * Do not cache a {@link DelegatingSessionFactory}, cache each delegate therein instead.
 	 * @see #setSessionWaitTimeout(long)
 	 * @see #setPoolSize(int)
 	 *
 	 * @param sessionFactory The underlying session factory.
 	 * @param sessionCacheSize The maximum cache size.
 	 */
 	public CachingSessionFactory(SessionFactory<F> sessionFactory, int sessionCacheSize) {
+		Assert.isTrue(!(sessionFactory instanceof DelegatingSessionFactory),
+				"'sessionFactory' cannot be a 'DelegatingSessionFactory'; cache each delegate instead");
 		this.sessionFactory = sessionFactory;
 		this.pool = new SimplePool<Session<F>>(sessionCacheSize, new SimplePool.PoolItemCallback<Session<F>>() {
 			@Override

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/DefaultSessionFactoryLocator.java

@@ -0,0 +1,78 @@
+/*
+ * Copyright 2015 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.integration.file.remote.session;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+/**
+ * The default implementation of {@link SessionFactoryLocator} using a simple map lookup
+ * and an optional default to fall back on.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class DefaultSessionFactoryLocator<F> implements SessionFactoryLocator<F> {
+
+	private final Map<Object, SessionFactory<F>> factories = new ConcurrentHashMap<Object, SessionFactory<F>>();
+
+	private final SessionFactory<F> defaultFactory;
+
+	/**
+	 * @param factories A map of factories, keyed by lookup key.
+	 */
+	public DefaultSessionFactoryLocator(Map<Object, SessionFactory<F>> factories) {
+		this(factories, null);
+	}
+
+	/**
+	 * @param factories A map of factories, keyed by lookup key.
+	 * @param defaultFactory A default to be used if the lookup fails.
+	 */
+	public DefaultSessionFactoryLocator(Map<Object, SessionFactory<F>> factories, SessionFactory<F> defaultFactory) {
+		this.factories.putAll(factories);
+		this.defaultFactory = defaultFactory;
+	}
+
+	/**
+	 * Add a session factory.
+	 * @param key the lookup key.
+	 * @param factory the factory.
+	 */
+	public void addSessionFactory(String key, SessionFactory<F> factory) {
+		this.factories.put(key, factory);
+	}
+
+	/**
+	 * Remove a session factory.
+	 * @param key the lookup key.
+	 * @return the factory, if it was present.
+	 */
+	public SessionFactory<F> removeSessionFactory(Object key) {
+		return this.factories.remove(key);
+	}
+
+	@Override
+	public SessionFactory<F> getSessionFactory(Object key) {
+		if (key == null) {
+			return this.defaultFactory;
+		}
+		SessionFactory<F> factory = this.factories.get(key);
+		return factory != null ? factory : this.defaultFactory;
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/DelegatingSessionFactory.java

@@ -0,0 +1,113 @@
+/*
+ * Copyright 2015 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.integration.file.remote.session;
+
+import java.util.Map;
+
+import org.springframework.messaging.Message;
+import org.springframework.util.Assert;
+
+/**
+ * {@link SessionFactory} that delegates to a {@link SessionFactory} retrieved from a
+ * {@link SessionFactoryLocator}.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public class DelegatingSessionFactory<F> implements SessionFactory<F> {
+
+	private final SessionFactoryLocator<F> factoryLocator;
+
+	private final ThreadLocal<Object> threadKey = new ThreadLocal<Object>();
+
+	/**
+	 * Construct an instance with a {@link DefaultSessionFactoryLocator} using the
+	 * supplied factories and default key.
+	 * @param factories the factories.
+	 * @param defaultFactory the default to use if the lookup fails.
+	 */
+	public DelegatingSessionFactory(Map<Object, SessionFactory<F>> factories, SessionFactory<F> defaultFactory) {
+		this(new DefaultSessionFactoryLocator<F>(factories, defaultFactory));
+	}
+
+	/**
+	 * Construct an instance using the supplied factory.
+	 * @param factoryLocator the factory.
+	 */
+	public DelegatingSessionFactory(SessionFactoryLocator<F> factoryLocator) {
+		Assert.notNull(factoryLocator, "'factoryFactory' cannot be null");
+		this.factoryLocator = factoryLocator;
+	}
+
+	/**
+	 * Return this factory's locator.
+	 * @return the locator.
+	 */
+	public SessionFactoryLocator<F> getFactoryLocator() {
+		return factoryLocator;
+	}
+
+	/**
+	 * Set a key to be used for {@link #getSession()} on this thread.
+	 * @param key the key.
+	 */
+	public void setThreadKey(Object key) {
+		this.threadKey.set(key);
+	}
+
+	/**
+	 * Clear the key for this thread.
+	 */
+	public void clearThreadKey() {
+		this.threadKey.remove();
+	}
+
+	/**
+	 * Messaging-friendly version of {@link #setThreadKey(Object)} that can be invoked from
+	 * a service activator.
+	 * @param message the message.
+	 * @param key the key.
+	 * @return the message (unchanged).
+	 */
+	public Message<?> setThreadKey(Message<?> message, Object key) {
+		this.threadKey.set(key);
+		return message;
+	}
+
+	/**
+	 * Messaging-friendly version of {@link #clearThreadKey()} that can be invoked from
+	 * a service activator.
+	 * @param message the message.
+	 * @return the message (unchanged).
+	 */
+	public Message<?> clearThreadKey(Message<?> message) {
+		this.threadKey.remove();
+		return message;
+	}
+
+	@Override
+	public Session<F> getSession() {
+		return getSession(this.threadKey.get());
+	}
+
+	public Session<F> getSession(Object key) {
+		SessionFactory<F> sessionFactory = this.factoryLocator.getSessionFactory(key);
+		Assert.notNull(sessionFactory, "No default SessionFactory configured");
+		return sessionFactory.getSession();
+	}
+
+}

spring-integration-file/src/main/java/org/springframework/integration/file/remote/session/SessionFactoryLocator.java

@@ -0,0 +1,35 @@
+/*
+ * Copyright 2015 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.integration.file.remote.session;
+
+/**
+ *
+ * A factory returning a {@link SessionFactory} based on some key.
+ *
+ * @author Gary Russell
+ * @since 4.2
+ *
+ */
+public interface SessionFactoryLocator<F> {
+
+	/**
+	 * Return a {@link SessionFactory} for the key.
+	 * @param key the key.
+	 * @return the session factory.
+	 */
+	SessionFactory<F> getSessionFactory(Object key);
+
+}