#207: Made the thread management more flexible by accepting a custom ExecutorService (ThreadPoolExecutor), with a default keepAliveTime of 1ms, allowing core threads to die idly as well

Author: bbottema

modules/core-module/src/main/java/org/simplejavamail/api/mailer/MailerGenericBuilder.java

@@ -12,6 +12,8 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.LinkedBlockingQueue;
 
 /**
  * Builder superclass which contains API to take care of all generic Mailer properties unrelated to the SMTP server
@@ -57,9 +59,9 @@ public interface MailerGenericBuilder<T extends MailerGenericBuilder<?>> {
 	boolean DEFAULT_JAVAXMAIL_DEBUG = false;
 
 	/**
-	 * Sets flag that send or server connection test should run in the background returning immediately.
+	 * Changes the default for sending emails and testing server connections to asynchronous (batch mode).
 	 * <p>
-	 * In case of asynchronous mode, make sure you configure logging to file, as otherwise you won't know if there was an error.
+	 * In case of asynchronous mode, make sure you configure logging to file or inspect the returned {@link AsyncResponse}.
 	 */
 	T async();
 
@@ -164,6 +166,27 @@ public interface MailerGenericBuilder<T extends MailerGenericBuilder<?>> {
 	 */
 	T withEmailAddressCriteria(@Nonnull EnumSet<EmailAddressCriteria> emailAddressCriteria);
 
+	/**
+	 * <strong>For advanced use cases.</strong>
+	 * <p>
+	 * Allows you to fully customize and manage the thread pool, threads and concurrency characteristics when
+	 * sending in batch mode.
+	 * <p>
+	 * By default the {@code NonJvmBlockingThreadPoolExecutor} is used:
+	 * <ul>
+	 *     <li>with core and max threads fixed to the given pool size</li>
+	 *     <li>with keepAliveTime as specified (if greater than zero, core thread will also time out and die off)</li>
+	 *     <li>A {@link LinkedBlockingQueue}</li>
+	 *     <li>The {@code NamedThreadFactory}, which creates named non-daemon threads</li>
+	 * </ul>
+	 * <p>
+	 * <strong>Note:</strong> What makes it NonJvm is that the default keepAliveTime is set to the lowest non-zero value (so 1), so that
+	 * any threads will die off as soon as possible, as not to block the JVM from shutting down.
+	 *
+	 * @param executorService A custom executor service (ThreadPoolExecutor), replacing the {@code NonJvmBlockingThreadPoolExecutor}.
+	 */
+	T withExecutorService(@Nonnull ExecutorService executorService);
+
 	/**
 	 * Sets both core thread pool size and max thread pool size to the given size.
 	 *
@@ -182,7 +205,7 @@ public interface MailerGenericBuilder<T extends MailerGenericBuilder<?>> {
 	 * When set to zero, this keepAliveTime is applied only to extra threads, not core threads. This is the classic executor
 	 * behavior, but this blocks the JVM from exiting.
 	 * <p>
-	 * Defaults to {@value #DEFAULT_POOL_KEEP_ALIVE_TIME}.
+	 * Defaults to {@value #DEFAULT_POOL_KEEP_ALIVE_TIME}ms.
 	 *
 	 * @param threadPoolKeepAliveTime Value in milliseconds. See main description for details.
 	 *
@@ -268,12 +291,17 @@ public interface MailerGenericBuilder<T extends MailerGenericBuilder<?>> {
 	 */
 	T resetEmailAddressCriteria();
 
+	/**
+	 * Resets the executor services to be used back to the default.
+	 *
+	 * @see #withExecutorService(ExecutorService)
+	 */
+	T resetExecutorService();
+
 	/**
 	 * Resets both thread pool max and core size to their defaults.
 	 *
 	 * @see #withThreadPoolSize(Integer)
-	 * @see #resetThreadpoolCoreSize()
-	 * @see #resetThreadpoolMaxSize()
 	 */
 	T resetThreadpoolSize();
 
@@ -373,6 +401,12 @@ public interface MailerGenericBuilder<T extends MailerGenericBuilder<?>> {
 	@Nullable
 	EnumSet<EmailAddressCriteria> getEmailAddressCriteria();
 
+	/**
+	 * @see #withExecutorService(ExecutorService)
+	 */
+	@Nullable
+	ExecutorService getExecutorService();
+
 	/**
 	 * @see #withThreadPoolSize(Integer)
 	 */

modules/core-module/src/main/java/org/simplejavamail/api/mailer/config/OperationalConfig.java

@@ -4,8 +4,10 @@
 import org.simplejavamail.api.mailer.MailerGenericBuilder;
 import org.simplejavamail.api.mailer.MailerRegularBuilder;
 
+import javax.annotation.Nonnull;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
 
 /**
  * Contains all the configuration that affect how a {@link Mailer} operates. This includes connection settings such as
@@ -47,6 +49,7 @@ public interface OperationalConfig {
 	/**
 	 * @see MailerRegularBuilder#trustingSSLHosts(String...)
 	 */
+	@Nonnull
 	List<String> getSslHostsToTrust();
 
 	/**
@@ -57,5 +60,12 @@ public interface OperationalConfig {
 	/**
 	 * @see MailerRegularBuilder#withProperties(Properties)
 	 */
+	@Nonnull
 	Properties getProperties();
+
+	/**
+	 * @see MailerRegularBuilder#withExecutorService(ExecutorService)
+	 */
+	@Nonnull
+	ExecutorService getExecutorService();
 }

modules/simple-java-mail/src/main/java/org/simplejavamail/mailer/internal/MailerGenericBuilderImpl.java

@@ -15,6 +15,7 @@
 import java.util.List;
 import java.util.Map;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
 
 import static org.simplejavamail.config.ConfigLoader.Property.PROXY_HOST;
 import static org.simplejavamail.config.ConfigLoader.Property.PROXY_PASSWORD;
@@ -80,6 +81,12 @@ public abstract class MailerGenericBuilderImpl<T extends MailerGenericBuilderImp
 	@Nonnull
 	private EnumSet<EmailAddressCriteria> emailAddressCriteria;
 
+	/**
+	 * @see MailerGenericBuilder#withExecutorService(ExecutorService)
+	 */
+	@Nullable
+	private ExecutorService executorService;
+
 	/**
 	 * @see MailerGenericBuilder#withThreadPoolSize(Integer)
 	 */
@@ -181,7 +188,8 @@ OperationalConfig buildOperationalConfig() {
 				isTransportModeLoggingOnly(),
 				isDebugLogging(),
 				getSslHostsToTrust(),
-				isTrustAllSSLHost());
+				isTrustAllSSLHost(),
+				getExecutorService());
 	}
 
 	/**
@@ -285,6 +293,15 @@ public T withEmailAddressCriteria(@Nonnull final EnumSet<EmailAddressCriteria> e
 		return (T) this;
 	}
 
+	/**
+	 * @see MailerGenericBuilder#withExecutorService(ExecutorService)
+	 */
+	@Override
+	public T withExecutorService(@Nonnull final ExecutorService executorService) {
+		this.executorService = executorService;
+		return (T) this;
+	}
+
 	/**
 	 * @see MailerGenericBuilder#withThreadPoolSize(Integer)
 	 */
@@ -381,6 +398,15 @@ public T resetEmailAddressCriteria() {
 		return withEmailAddressCriteria(EmailAddressCriteria.RFC_COMPLIANT);
 	}
 
+	/**
+	 * @see MailerGenericBuilder#resetExecutorService()
+	 */
+	@Override
+	public T resetExecutorService() {
+		this.executorService = null;
+		return (T) this;
+	}
+
 	/**
 	 * @see MailerGenericBuilder#resetThreadpoolSize()
 	 */
@@ -518,6 +544,15 @@ public EnumSet<EmailAddressCriteria> getEmailAddressCriteria() {
 		return emailAddressCriteria;
 	}
 
+	/**
+	 * @see MailerGenericBuilder#getExecutorService()
+	 */
+	@Override
+	@Nullable
+	public ExecutorService getExecutorService() {
+		return executorService;
+	}
+
 	/**
 	 * @see MailerGenericBuilder#getThreadPoolSize()
 	 */

modules/simple-java-mail/src/main/java/org/simplejavamail/mailer/internal/OperationalConfigImpl.java

@@ -2,14 +2,19 @@
 
 import org.simplejavamail.api.mailer.MailerRegularBuilder;
 import org.simplejavamail.api.mailer.config.OperationalConfig;
+import org.simplejavamail.mailer.internal.mailsender.concurrent.NonJvmBlockingThreadPoolExecutor;
 
+import javax.annotation.Nonnull;
+import javax.annotation.Nullable;
 import java.util.Collections;
 import java.util.List;
 import java.util.Properties;
+import java.util.concurrent.ExecutorService;
 
 /**
  * @see OperationalConfig
  */
+// FIXME Lombok
 public class OperationalConfigImpl implements OperationalConfig {
 	/**
 	 * @see MailerRegularBuilder#withSessionTimeout(Integer)
@@ -48,20 +53,27 @@ public class OperationalConfigImpl implements OperationalConfig {
 	/**
 	 * @see MailerRegularBuilder#trustingSSLHosts(String...)
 	 */
+	@Nonnull
 	private final List<String> sslHostsToTrust;
-	
+
 	/**
 	 * @see MailerRegularBuilder#trustingAllHosts(boolean)
 	 */
 	private final boolean trustAllSSLHost;
+
+	/**
+	 * @see MailerRegularBuilder#withExecutorService(ExecutorService)
+	 */
+	@Nonnull
+	private final ExecutorService executorService;
 
 	/**
 	 * @deprecated For internal use only.
 	 */
 	@Deprecated
 	@SuppressWarnings("DeprecatedIsStillUsed")
-	public OperationalConfigImpl(boolean async, Properties properties, int sessionTimeout, int threadPoolSize, int threadPoolKeepAliveTime, boolean transportModeLoggingOnly, boolean debugLogging,
-			List<String> sslHostsToTrust, boolean trustAllSSLHost) {
+	public OperationalConfigImpl(final boolean async, Properties properties, int sessionTimeout, int threadPoolSize, int threadPoolKeepAliveTime, boolean transportModeLoggingOnly,
+			boolean debugLogging, @Nonnull List<String> sslHostsToTrust, boolean trustAllSSLHost, @Nullable final ExecutorService executorService) {
 		this.async = async;
 		this.properties = properties;
 		this.sessionTimeout = sessionTimeout;
@@ -71,8 +83,11 @@ public OperationalConfigImpl(boolean async, Properties properties, int sessionTi
 		this.debugLogging = debugLogging;
 		this.sslHostsToTrust = Collections.unmodifiableList(sslHostsToTrust);
 		this.trustAllSSLHost = trustAllSSLHost;
+		this.executorService = executorService != null
+				? executorService
+				: new NonJvmBlockingThreadPoolExecutor(getThreadPoolSize(), getThreadPoolKeepAliveTime());
 	}
-	
+
 	/**
 	 * @see OperationalConfig#isAsync()
 	 */
@@ -124,6 +139,7 @@ public boolean isDebugLogging() {
 	/**
 	 * @see OperationalConfig#getSslHostsToTrust()
 	 */
+	@Nonnull
 	@Override
 	public List<String> getSslHostsToTrust() {
 		return sslHostsToTrust;
@@ -140,8 +156,15 @@ public boolean isTrustAllSSLHost() {
 	/**
 	 * @see OperationalConfig#getProperties()
 	 */
+	@Nonnull
 	@Override
 	public Properties getProperties() {
 		return properties;
 	}
+
+	@Nonnull
+	@Override
+	public ExecutorService getExecutorService() {
+		return executorService;
+	}
 }

modules/simple-java-mail/src/main/java/org/simplejavamail/mailer/internal/mailsender/MailSenderImpl.java

@@ -24,7 +24,6 @@
 import java.io.UnsupportedEncodingException;
 import java.util.List;
 import java.util.Properties;
-import java.util.concurrent.ExecutorService;
 import java.util.concurrent.Phaser;
 
 import static java.lang.String.format;
@@ -49,6 +48,7 @@ public class MailSenderImpl implements MailSender {
 	/**
 	 * Used to actually send the email. This session can come from being passed in the default constructor, or made by <code>Mailer</code> directly.
 	 */
+	@Nonnull
 	private final Session session;
 
 	/**
@@ -65,6 +65,7 @@ public class MailSenderImpl implements MailSender {
 	/**
 	 * @see OperationalConfig
 	 */
+	@Nonnull
 	private final OperationalConfig operationalConfig;
 
 	/**
@@ -74,14 +75,6 @@ public class MailSenderImpl implements MailSender {
 	@Nullable
 	private final AnonymousSocks5Server proxyServer;
 
-	/**
-	 * Allows us to manage how many thread we run at the same time using a thread pool.
-	 * <p>
-	 * Can't be initialized in the field, because we need to reinitialize it whenever the threadpool was closed after a batch of emails and this
-	 * MailSender instance is again engaged.
-	 */
-	private ExecutorService executor;
-
 	/**
 	 * Used to keep track of running SMTP requests, so that we know when to close down the proxy bridging server (if used).
 	 * <p>
@@ -106,7 +99,7 @@ private void initSession(@Nonnull final Session session, @Nonnull OperationalCon
 		session.setDebug(operationalConfig.isDebugLogging());
 		session.getProperties().putAll(operationalConfig.getProperties());
 
-		configureSessionWithTimeout(session, operationalConfig.getSessionTimeout());
+		configureSessionWithTimeout(session, operationalConfig.getSessionTimeout(), transportStrategy);
 
 		if (transportStrategy != null) {
 			if (operationalConfig.isTrustAllSSLHost()) {
@@ -186,18 +179,14 @@ the proxy bridge server (or connection pool in async mode) while a non-async ema
 			sendMailClosure.run();
 			return null;
 		} else {
-			// start up thread pool if necessary
-			if (executor == null) {
-				executor = new NonJvmBlockingThreadPoolExecutor(operationalConfig, "Simple Java Mail async mail sender");
-			}
-			return AsyncOperationHelper.executeAsync(executor, "sendMail process", sendMailClosure);
+			return AsyncOperationHelper.executeAsync(operationalConfig.getExecutorService(), "sendMail process", sendMailClosure);
 		}
 	}
 
 	/**
 	 * Configures the {@link Session} with the same timeout for socket connection timeout, read and write timeout.
 	 */
-	private void configureSessionWithTimeout(final Session session, final int sessionTimeout) {
+	private void configureSessionWithTimeout(@Nonnull final Session session, final int sessionTimeout, @Nullable final TransportStrategy transportStrategy) {
 		if (transportStrategy != null) {
 			// socket timeouts handling
 			final Properties sessionProperties = session.getProperties();
@@ -384,7 +373,7 @@ public void run() {
 
 			boolean proxyBridgeStartedForTestingConnection = false;
 
-			configureSessionWithTimeout(session, operationalConfig.getSessionTimeout());
+			configureSessionWithTimeout(session, operationalConfig.getSessionTimeout(), transportStrategy);
 
 			logSession(session, async, "connection test");
 
@@ -419,6 +408,7 @@ private boolean needsAuthenticatedProxy() {
 	 * @see MailSender#getSession()
 	 */
 	@Override
+	@Nonnull
 	public Session getSession() {
 		return session;
 	}
@@ -427,6 +417,7 @@ public Session getSession() {
 	 * @see MailSender#getOperationalConfig()
 	 */
 	@Override
+	@Nonnull
 	public OperationalConfig getOperationalConfig() {
 		return operationalConfig;
 	}