#198: revamp

Author: bbottema

modules/batch-module/src/main/java/org/simplejavamail/internal/batchsupport/BatchException.java

@@ -0,0 +1,10 @@
+package org.simplejavamail.internal.batchsupport;
+
+class BatchException extends RuntimeException {
+
+	static final String ERROR_ACQUIRING_KEYED_POOLABLE = "Was unable to obtain a poolable object for key:\t\n%s";
+
+	BatchException(final String msg, final Throwable cause) {
+		super(msg, cause);
+	}
+}

modules/batch-module/src/main/java/org/simplejavamail/internal/batchsupport/BatchSupport.java

@@ -5,7 +5,7 @@
 import org.simplejavamail.internal.batchsupport.concurrent.NonJvmBlockingThreadPoolExecutor;
 import org.simplejavamail.internal.batchsupport.transportpool.LifecycleDelegatingTransportImpl;
 import org.simplejavamail.internal.batchsupport.transportpool.PoolableTransportAllocatorFactory;
-import org.simplejavamail.internal.batchsupport.transportpool.keyedcloseablepools.KeyedObjectPools;
+import org.simplejavamail.internal.batchsupport.transportpool.keyedcloseablepools.KeyedCyclingObjectPools;
 import org.simplejavamail.internal.batchsupport.transportpool.keyedcloseablepools.SimpleDelegatingPoolable;
 import org.simplejavamail.internal.modules.BatchModule;
 import org.slf4j.Logger;
@@ -19,7 +19,9 @@
 import java.util.concurrent.ExecutorService;
 import java.util.concurrent.TimeUnit;
 
+import static java.lang.String.format;
 import static java.util.concurrent.TimeUnit.SECONDS;
+import static org.simplejavamail.internal.batchsupport.BatchException.ERROR_ACQUIRING_KEYED_POOLABLE;
 
 /**
  * This class only serves to hide the Batch implementation behind an easy-to-load-with-reflection class.
@@ -31,8 +33,8 @@ public class BatchSupport implements BatchModule {
 	private static final Timeout WAIT_FOREVER = new Timeout(Long.MAX_VALUE, TimeUnit.DAYS);
 	private static final TimeExpiration<SimpleDelegatingPoolable<Transport>> TIME_TO_LIVE_5_SECONDS = new TimeExpiration<>(5, SECONDS);
 
-	private final KeyedObjectPools<Session, SimpleDelegatingPoolable<Transport>> transportPools =
-			new KeyedObjectPools<>(new PoolableTransportAllocatorFactory(), TIME_TO_LIVE_5_SECONDS, WAIT_FOREVER);
+	private final KeyedCyclingObjectPools<Session, SimpleDelegatingPoolable<Transport>> transportPools =
+			new KeyedCyclingObjectPools<>(new PoolableTransportAllocatorFactory(), TIME_TO_LIVE_5_SECONDS, WAIT_FOREVER);
 
 	/**
 	 * @see BatchModule#executeAsync(String, Runnable)
@@ -68,7 +70,11 @@ public ExecutorService createDefaultExecutorService(final int threadPoolSize, fi
 	@Override
 	@SuppressWarnings("deprecation")
 	public LifecycleDelegatingTransport acquireTransport(@Nonnull final Session session) {
-		return new LifecycleDelegatingTransportImpl(transportPools.acquire(session));
+		try {
+			return new LifecycleDelegatingTransportImpl(transportPools.acquire(session));
+		} catch (InterruptedException e) {
+			throw new BatchException(format(ERROR_ACQUIRING_KEYED_POOLABLE, session), e);
+		}
 	}
 
 	/**

modules/batch-module/src/main/java/org/simplejavamail/internal/batchsupport/transportpool/keyedcloseablepools/AcquireKeyedPoolableException.java

@@ -0,0 +1,74 @@
+package org.simplejavamail.internal.batchsupport.transportpool.keyedcloseablepools;
+
+import stormpot.BlazePool;
+import stormpot.Config;
+import stormpot.Expiration;
+import stormpot.LifecycledPool;
+import stormpot.Poolable;
+import stormpot.Timeout;
+
+import javax.annotation.Nonnull;
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.ConcurrentHashMap;
+
+import static java.util.Collections.singletonList;
+
+/**
+ * <ol>
+ *     <li>Keyed: because you govern object pools based on keys, like a cache on steroids</li>
+ *     <li>Cycling: because you can cycle multiple pools per key round robin</li>
+ * </ol>
+ * Example usage:
+ * <p>
+ * Say you have two different mail clusters, each with several servers. The keys map to the clusters, where each server
+ * is accessed round robin, and the objects in the respective cluster-pool are multiple connections to the same mail server.
+ * <p>
+ * FIXME implement key properly and then document
+ */
+public class KeyedCyclingObjectPools<Key, T extends Poolable> {
+
+	private final ConcurrentHashMap<Key, Queue<LifecycledPool<T>>> keyedPool = new ConcurrentHashMap<>();
+
+	@Nonnull private final AllocatorFactory<Key, T> allocatorFactory;
+	@Nonnull private final Expiration<T> expirationPolicy;
+	@Nonnull private final Timeout claimTimeout;
+
+	public KeyedCyclingObjectPools(
+			@Nonnull final AllocatorFactory<Key, T> allocatorFactory,
+			@Nonnull final Expiration<T> expirationPolicy,
+			@Nonnull final Timeout claimTimeout) {
+		this.allocatorFactory = allocatorFactory;
+		this.expirationPolicy = expirationPolicy;
+		this.claimTimeout = claimTimeout;
+	}
+
+	public T acquire(final Key key) throws InterruptedException {
+		return findPool(key).claim(claimTimeout);
+	}
+
+	/**
+	 * Clearing a pool is like shutting down a pool, but semantically speaking, a cleared pool can be restarted again.
+	 */
+	public synchronized void clearPool(final Key key) {
+		if (keyedPool.containsKey(key)) {
+			for (final LifecycledPool<T> poolInCluster : keyedPool.remove(key)) {
+				poolInCluster.shutdown();
+			}
+		}
+	}
+
+	private synchronized LifecycledPool<T> findPool(final Key key) {
+		if (!keyedPool.containsKey(key)) {
+			LifecycledPool<T> pool = new BlazePool<>(new Config<T>()
+					.setExpiration(expirationPolicy)
+					// FIXME how to create a cluster?
+					.setAllocator(allocatorFactory.create(key)));
+			keyedPool.put(key, new LinkedList<>(singletonList(pool)));
+		}
+		Queue<LifecycledPool<T>> lifecycledPools = keyedPool.get(key);
+		LifecycledPool<T> nextRoundrobinInCluster = lifecycledPools.remove();
+		lifecycledPools.add(nextRoundrobinInCluster);
+		return nextRoundrobinInCluster;
+	}
+}