Fix #915 by adding ExecutorServiceProvider to enable developers to use their own ExecutorService (#916)

* Fix #915 by adding ExecutorServiceProvider to enable developers to use their own ExecutorService
* Fix broken tests
* Improve the metrics database code
* Fix tests

Author: seratch

bolt-servlet/src/test/java/test_with_remote_apis/sample_json_generation/JsonDataRecordingListener.java

@@ -1,7 +1,7 @@
 package test_with_remote_apis.sample_json_generation;
 
 import com.slack.api.util.http.listener.HttpResponseListener;
-import com.slack.api.util.thread.ExecutorServiceFactory;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
 import lombok.extern.slf4j.Slf4j;
 
 import java.io.IOException;
@@ -15,7 +15,8 @@ public class JsonDataRecordingListener extends HttpResponseListener {
 
     private final CopyOnWriteArrayList<String> remaining = new CopyOnWriteArrayList<>();
     private final String threadGroupName = "slack-unit-test-JsonDataRecordingListener";
-    private final ExecutorService executorService = ExecutorServiceFactory.createDaemonThreadPoolExecutor(threadGroupName, 5);
+    private final ExecutorService executorService = DaemonThreadExecutorServiceProvider.getInstance()
+            .createThreadPoolExecutor(threadGroupName, 5);
 
     public boolean isAllDone() {
         if (remaining.size() > 0) {

bolt-socket-mode/src/test/java/util/socket_mode/MockSocketMode.java

@@ -1,6 +1,6 @@
 package util.socket_mode;
 
-import com.slack.api.util.thread.ExecutorServiceFactory;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
 import lombok.extern.slf4j.Slf4j;
 import org.eclipse.jetty.websocket.api.Session;
 import org.eclipse.jetty.websocket.api.StatusCode;
@@ -22,7 +22,7 @@ public class MockSocketMode extends WebSocketAdapter {
     private CountDownLatch closureLatch = new CountDownLatch(1);
 
     private CopyOnWriteArrayList<Session> activeSessions = new CopyOnWriteArrayList<>();
-    private ScheduledExecutorService service = ExecutorServiceFactory.createDaemonThreadScheduledExecutor(
+    private ScheduledExecutorService service = DaemonThreadExecutorServiceProvider.getInstance().createThreadScheduledExecutor(
             MockSocketMode.class.getCanonicalName());
 
     public MockSocketMode() {

bolt/src/main/java/com/slack/api/bolt/App.java

@@ -2,6 +2,7 @@
 
 import com.google.gson.Gson;
 import com.slack.api.Slack;
+import com.slack.api.SlackConfig;
 import com.slack.api.app_backend.SlackSignature;
 import com.slack.api.app_backend.events.EventHandler;
 import com.slack.api.app_backend.events.EventsDispatcher;
@@ -32,7 +33,6 @@
 import com.slack.api.model.event.MessageEvent;
 import com.slack.api.model.event.TokensRevokedEvent;
 import com.slack.api.util.json.GsonFactory;
-import com.slack.api.util.thread.ExecutorServiceFactory;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.extern.slf4j.Slf4j;
@@ -440,15 +440,29 @@ public App(AppConfig appConfig) {
     }
 
     public App(AppConfig appConfig, List<Middleware> middlewareList) {
-        this(appConfig, appConfig.getSlack() != null ? appConfig.getSlack() : Slack.getInstance(), middlewareList);
+        this(appConfig, appConfig.getSlack() != null
+                        ? appConfig.getSlack()
+                        // Intentionally instantiating a new SlackConfig instance
+                        // to avoid using the immutable singleton one
+                        : Slack.getInstance(new SlackConfig()),
+                middlewareList
+        );
         this.status = Status.Stopped;
     }
 
     public App(AppConfig appConfig, Slack slack, List<Middleware> middlewareList) {
         this.appConfig = appConfig;
-        this.executorService = ExecutorServiceFactory.createDaemonThreadPoolExecutor(
+        this.appConfig.setSlack(slack);
+        SlackConfig clientConfig = this.appConfig.getSlack().getConfig();
+        if (!clientConfig.getExecutorServiceProvider().equals(this.appConfig.getExecutorServiceProvider())) {
+            clientConfig.setExecutorServiceProvider(this.appConfig.getExecutorServiceProvider());
+            clientConfig.synchronizeExecutorServiceProviders();
+        }
+
+        this.executorService = this.appConfig.getExecutorServiceProvider().createThreadPoolExecutor(
                 "bolt-app-threads",
-                this.appConfig.getThreadPoolSize());
+                this.appConfig.getThreadPoolSize()
+        );
         this.middlewareList = middlewareList;
 
         this.oAuthStateService = new ClientOnlyOAuthStateService();

bolt/src/main/java/com/slack/api/bolt/AppConfig.java

@@ -11,6 +11,8 @@
 import com.slack.api.bolt.service.builtin.oauth.view.default_impl.OAuthDefaultRedirectUriPageRenderer;
 import com.slack.api.token_rotation.TokenRotator;
 import com.slack.api.util.http.SlackHttpClient;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
+import com.slack.api.util.thread.ExecutorServiceProvider;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
@@ -67,8 +69,12 @@ private EnvVariableName() {
         public static final String SLACK_APP_OAUTH_COMPLETION_URL = "SLACK_APP_OAUTH_COMPLETION_URL";
     }
 
+    // We don't use SlackConfig.DEFAULT here to enable developers to customize the properties later on
     @Builder.Default
-    private transient Slack slack = Slack.getInstance(SlackConfig.DEFAULT, buildSlackHttpClient());
+    private transient Slack slack = Slack.getInstance(new SlackConfig(), buildSlackHttpClient());
+
+    @Builder.Default
+    private transient ExecutorServiceProvider executorServiceProvider = DaemonThreadExecutorServiceProvider.getInstance();
 
     private static SlackHttpClient buildSlackHttpClient() {
         Map<String, String> userAgentCustomInfo = new HashMap<>();

bolt/src/main/java/com/slack/api/bolt/middleware/builtin/MultiTeamsAuthorization.java

@@ -98,7 +98,7 @@ public MultiTeamsAuthorization(AppConfig config, InstallationService installatio
     private ScheduledExecutorService buildTokenToAuthTestCacheCleaner(Runnable task) {
         String threadGroupName = MultiTeamsAuthorization.class.getSimpleName();
         ScheduledExecutorService tokenToAuthTestCacheCleaner =
-                ExecutorServiceFactory.createDaemonThreadScheduledExecutor(threadGroupName);
+                this.config.getExecutorServiceProvider().createThreadScheduledExecutor(threadGroupName);
         tokenToAuthTestCacheCleaner.scheduleAtFixedRate(task, 120_000, 30_000, TimeUnit.MILLISECONDS);
         log.debug("The tokenToAuthTestCacheCleaner (daemon thread) started");
         return tokenToAuthTestCacheCleaner;

bolt/src/main/java/com/slack/api/bolt/middleware/builtin/WorkflowStep.java

@@ -10,7 +10,7 @@
 import com.slack.api.bolt.request.builtin.WorkflowStepExecuteRequest;
 import com.slack.api.bolt.request.builtin.WorkflowStepSaveRequest;
 import com.slack.api.bolt.response.Response;
-import com.slack.api.util.thread.ExecutorServiceFactory;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
 import lombok.*;
 import lombok.extern.slf4j.Slf4j;
 
@@ -30,8 +30,13 @@ public class WorkflowStep implements Middleware, AutoCloseable {
     private WorkflowStepEditHandler edit;
     private WorkflowStepSaveHandler save;
     private WorkflowStepExecuteHandler execute;
+
     @Builder.Default
     private boolean executeAutoAcknowledgement = true;
+
+    // If a developer would like to use their own ExecutorService here,
+    // the recommended way would be to pass the one using the builder method:
+    // `WorkflowStep.builder().executorService(executorService).build()`
     @Builder.Default
     private ExecutorService executorService = buildDefaultExecutorService();
 
@@ -94,7 +99,7 @@ public Response apply(Request req, Response resp, MiddlewareChain chain) throws
                                 // Auto acknowledgement
                                 return new Response();
                             } else {
-                                // In the execute handler, workflows.stepCompleted/stepFailed
+                                // In the `execute` handler, workflows.stepCompleted/stepFailed
                                 // needs to be called asynchronously
                                 return execute.apply(request, request.getContext());
                             }
@@ -125,7 +130,9 @@ && getCallbackIdPattern().matcher(requestCallbackId).matches())
     protected static ExecutorService buildDefaultExecutorService() {
         String threadGroupName = WorkflowStep.class.getSimpleName();
         int poolSize = 3;
-        ExecutorService service = ExecutorServiceFactory.createDaemonThreadPoolExecutor(
+        // If you want to use own ExecutorService, pass it using the builder method instead:
+        // `WorkflowStep.builder().executorService(executorService).build()`
+        ExecutorService service = DaemonThreadExecutorServiceProvider.getInstance().createThreadPoolExecutor(
                 threadGroupName,
                 poolSize
         );

bolt/src/test/java/test_locally/AppTest.java

@@ -15,12 +15,16 @@
 import com.slack.api.bolt.service.builtin.oauth.OAuthV2SuccessHandler;
 import com.slack.api.bolt.service.builtin.oauth.default_impl.OAuthDefaultExceptionHandler;
 import com.slack.api.bolt.service.builtin.oauth.default_impl.OAuthDefaultStateErrorHandler;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceFactory;
+import com.slack.api.util.thread.ExecutorServiceProvider;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.After;
 import org.junit.Before;
 import org.junit.Test;
 import util.AuthTestMockServer;
 
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.ScheduledExecutorService;
 import java.util.concurrent.atomic.AtomicBoolean;
 
 import static org.hamcrest.CoreMatchers.is;
@@ -358,4 +362,29 @@ public void clientConfig() {
                 is("http://localhost:8080/new"));
     }
 
+    @Test
+    public void customExecutorService() {
+        final AtomicBoolean called = new AtomicBoolean(false);
+        ExecutorServiceProvider executorServiceProvider = new ExecutorServiceProvider() {
+            @Override
+            public ExecutorService createThreadPoolExecutor(String threadGroupName, int poolSize) {
+                called.set(true);
+                return DaemonThreadExecutorServiceFactory.createDaemonThreadPoolExecutor(
+                        threadGroupName, poolSize);
+            }
+
+            @Override
+            public ScheduledExecutorService createThreadScheduledExecutor(String threadGroupName) {
+                return DaemonThreadExecutorServiceFactory.createDaemonThreadScheduledExecutor(threadGroupName);
+            }
+        };
+        App app = new App(AppConfig.builder()
+                .signingSecret("secret")
+                .singleTeamBotToken("token")
+                .executorServiceProvider(executorServiceProvider)
+                .build());
+        assertThat(app.config().getExecutorServiceProvider(), is(executorServiceProvider));
+        assertThat(called.get(), is(true));
+    }
+
 }

bolt/src/test/java/test_locally/app/WorkflowStepTest.java

@@ -10,6 +10,7 @@
 import com.slack.api.bolt.request.builtin.WorkflowStepEditRequest;
 import com.slack.api.bolt.request.builtin.WorkflowStepSaveRequest;
 import com.slack.api.bolt.response.Response;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
 import lombok.extern.slf4j.Slf4j;
 import org.junit.After;
 import org.junit.Before;
@@ -18,7 +19,11 @@
 
 import java.net.URLEncoder;
 import java.util.*;
+import java.util.concurrent.ExecutorService;
 
+import static org.hamcrest.CoreMatchers.is;
+import static org.hamcrest.CoreMatchers.notNullValue;
+import static org.hamcrest.MatcherAssert.assertThat;
 import static org.junit.Assert.assertEquals;
 
 @Slf4j
@@ -317,4 +322,13 @@ void setRequestHeaders(String requestBody, Map<String, List<String>> rawHeaders,
         rawHeaders.put(SlackSignature.HeaderNames.X_SLACK_SIGNATURE, Arrays.asList(generator.generate(timestamp, requestBody)));
     }
 
+    @Test
+    public void buildWithCustomExecutorService() {
+        ExecutorService executorService = DaemonThreadExecutorServiceProvider.getInstance()
+                .createThreadPoolExecutor("foo", 3);
+        WorkflowStep step = WorkflowStep.builder().executorService(executorService).build();
+        assertThat(step, is(notNullValue()));
+        assertThat(step.getExecutorService(), is(executorService));
+    }
+
 }

slack-api-client/src/main/java/com/slack/api/Slack.java

@@ -77,6 +77,9 @@ private Slack(SlackConfig config) {
 
     private Slack(SlackConfig config, SlackHttpClient httpClient) {
         this.config = config;
+        if (!this.config.equals(SlackConfig.DEFAULT)) {
+            this.config.synchronizeExecutorServiceProviders();
+        }
         this.httpClient = httpClient;
         this.httpClient.setConfig(this.config);
     }

slack-api-client/src/main/java/com/slack/api/SlackConfig.java

@@ -11,6 +11,9 @@
 import com.slack.api.util.http.listener.DetailedLoggingListener;
 import com.slack.api.util.http.listener.HttpResponseListener;
 import com.slack.api.util.http.listener.ResponsePrettyPrintingListener;
+import com.slack.api.util.thread.DaemonThreadExecutorServiceProvider;
+import com.slack.api.util.thread.ExecutorServiceProvider;
+import lombok.Builder;
 import lombok.Data;
 
 import java.util.ArrayList;
@@ -121,6 +124,11 @@ public void setHttpClientWriteTimeoutMillis(Integer httpClientWriteTimeoutMillis
         public void setHttpClientReadTimeoutMillis(Integer httpClientReadTimeoutMillis) {
             throwException();
         }
+
+        @Override
+        public void setExecutorServiceProvider(ExecutorServiceProvider executorServiceProvider) {
+            throwException();
+        }
     };
 
     public SlackConfig() {
@@ -213,9 +221,30 @@ private static String initProxyUrl() {
 
     private String legacyStatusEndpointUrlPrefix = LegacyStatusClient.ENDPOINT_URL_PREFIX;
 
-    private MethodsConfig methodsConfig = MethodsConfig.DEFAULT_SINGLETON;
+    @Builder.Default
+    private ExecutorServiceProvider executorServiceProvider = DaemonThreadExecutorServiceProvider.getInstance();
+
+    private MethodsConfig methodsConfig = new MethodsConfig();
 
-    private AuditConfig auditConfig = AuditConfig.DEFAULT_SINGLETON;
+    private AuditConfig auditConfig = new AuditConfig();
 
-    private SCIMConfig sCIMConfig = SCIMConfig.DEFAULT_SINGLETON;
+    private SCIMConfig sCIMConfig = new SCIMConfig();
+
+    public void synchronizeExecutorServiceProviders() {
+        if (!methodsConfig.equals(MethodsConfig.DEFAULT_SINGLETON)
+                && !methodsConfig.getExecutorServiceProvider().equals(executorServiceProvider)) {
+            methodsConfig.setExecutorServiceProvider(executorServiceProvider);
+            methodsConfig.getMetricsDatastore().setExecutorServiceProvider(executorServiceProvider);
+        }
+        if (!auditConfig.equals(AuditConfig.DEFAULT_SINGLETON)
+                && !auditConfig.getExecutorServiceProvider().equals(executorServiceProvider)) {
+            auditConfig.setExecutorServiceProvider(executorServiceProvider);
+            auditConfig.getMetricsDatastore().setExecutorServiceProvider(executorServiceProvider);
+        }
+        if (!sCIMConfig.equals(SCIMConfig.DEFAULT_SINGLETON)
+                && !sCIMConfig.getExecutorServiceProvider().equals(executorServiceProvider)) {
+            sCIMConfig.setExecutorServiceProvider(executorServiceProvider);
+            sCIMConfig.getMetricsDatastore().setExecutorServiceProvider(executorServiceProvider);
+        }
+    }
 }