RCB-597: move thread pool to example

Author: Adam Soos

api/src/main/java/com/basistech/rosette/api/HttpRosetteAPI.java

@@ -67,18 +67,13 @@
 import java.net.URISyntaxException;
 import java.nio.charset.StandardCharsets;
 import java.util.ArrayList;
-import java.util.Collection;
 import java.util.Collections;
 import java.util.HashSet;
 import java.util.List;
 import java.util.Properties;
 import java.util.Set;
-import java.util.concurrent.ExecutorService;
-import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
-import java.util.concurrent.TimeUnit;
 import java.util.regex.Pattern;
-import java.util.stream.Collectors;
 import java.util.zip.GZIPInputStream;
 
 import static java.net.HttpURLConnection.HTTP_OK;
@@ -103,8 +98,6 @@ public class HttpRosetteAPI extends AbstractRosetteAPI {
     private List<Header> additionalHeaders;
     private int connectionConcurrency = 2;
     private boolean closeClientOnClose = true;
-    private ExecutorService threadPool;
-    private int shutdownWait;
 
     private HttpRosetteAPI() {
         // use builder
@@ -126,8 +119,7 @@ private HttpRosetteAPI() {
      */
     HttpRosetteAPI(String key, String urlToCall, Integer failureRetries,
                        CloseableHttpClient httpClient, List<Header> additionalHeaders,
-                       Integer connectionConcurrency, boolean onlyAcceptKnownFields,
-                       Integer shutdownWait) throws HttpRosetteAPIException {
+                       Integer connectionConcurrency, boolean onlyAcceptKnownFields) throws HttpRosetteAPIException {
         urlBase = urlToCall == null ? urlBase : TRAILING_SLASHES.matcher(urlToCall.trim()).replaceAll("");
         if (failureRetries != null && failureRetries >= 1) {
             this.failureRetries = failureRetries;
@@ -136,10 +128,6 @@ private HttpRosetteAPI() {
         if (connectionConcurrency != null) {
             this.connectionConcurrency = connectionConcurrency;
         }
-        this.threadPool = Executors.newFixedThreadPool(this.connectionConcurrency * 2);
-        if (shutdownWait != null) {
-            this.shutdownWait = shutdownWait;
-        }
 
         mapper = ApiModelMixinModule.setupObjectMapper(new ObjectMapper());
         mapper.configure(DeserializationFeature.FAIL_ON_UNKNOWN_PROPERTIES, onlyAcceptKnownFields);
@@ -627,41 +615,12 @@ public RosetteRequest createRosetteRequest(String endpoint,
         return new RosetteRequest(this, request, endpoint, responseClass);
     }
 
-    /**
-     * Sends the request concurrently
-     * @param request the request to be sent
-     * @return A Future with the response of the request
-     */
-    public Future<Response> submitRequest(RosetteRequest request) {
-        return this.threadPool.submit(request);
-    }
-
-    /**
-     * Sends the requests concurrently
-     * @param requests list of the requests to be sent
-     * @return A list of Futures with the responses to the requests
-     */
-    public List<Future<Response>> submitRequests(Collection<RosetteRequest> requests) {
-        return requests.stream()
-                .map(this::submitRequest)
-                .collect(Collectors.toList());
-    }
-
 
     @Override
     public void close() throws IOException {
         if (closeClientOnClose) {
             httpClient.close();
         }
-        try {
-            this.threadPool.shutdown();
-            boolean terminated = this.threadPool.awaitTermination(this.shutdownWait, TimeUnit.SECONDS);
-            if (!terminated) {
-                this.threadPool.shutdownNow();
-            }
-        } catch (InterruptedException e) {
-            this.threadPool.shutdownNow();
-        }
     }
 
     /**
@@ -675,7 +634,6 @@ public static class Builder {
         private CloseableHttpClient httpClient;
         private List<Header> additionalHeaders = new ArrayList<>();
         private boolean onlyAcceptKnownFields;
-        private Integer shutdownWait;
 
         /**
          * Specify the API key. This is required for use with the public API, and
@@ -758,16 +716,6 @@ public Builder onlyAcceptKnownFields(boolean onlyAcceptKnownFields) {
             return this;
         }
 
-        /**
-         * Specify how long in seconds the api should wait for pending requests before forcefully terminating
-         * @param shutdownWaitSeconds the length of the wait in seconds. The default is 0.
-         * @return this.
-         */
-        public Builder shutdownWait(int shutdownWaitSeconds) {
-            this.shutdownWait = shutdownWaitSeconds;
-            return this;
-        }
-
         /**
          * Build the API object.
          * @return the new API object.
@@ -780,8 +728,7 @@ public HttpRosetteAPI build() throws HttpRosetteAPIException {
                                       httpClient,
                                       additionalHeaders,
                                       concurrency,
-                                      onlyAcceptKnownFields,
-                                      shutdownWait);
+                                      onlyAcceptKnownFields);
         }
     }
 }

api/src/test/java/com/basistech/rosette/api/RosetteRequestTest.java

@@ -23,8 +23,6 @@
 import com.basistech.rosette.apimodel.LanguageOptions;
 import com.basistech.rosette.apimodel.LanguageResponse;
 import com.basistech.rosette.apimodel.Response;
-import org.apache.http.impl.client.CloseableHttpClient;
-import org.apache.http.impl.client.HttpClients;
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
@@ -40,12 +38,13 @@
 import java.util.Date;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 import java.util.concurrent.TimeUnit;
 
 import static org.junit.jupiter.api.Assertions.assertEquals;
 import static org.junit.jupiter.api.Assertions.assertInstanceOf;
-import static org.junit.jupiter.api.Assertions.assertNull;
 import static org.junit.jupiter.api.Assertions.assertTrue;
 
 @ExtendWith(MockServerExtension.class)
@@ -86,9 +85,11 @@ void successfulRequest() throws ExecutionException, InterruptedException {
         RosetteRequest entitiesRequest = this.api.createRosetteRequest(AbstractRosetteAPI.ENTITIES_SERVICE_PATH, entitiesRequestData, EntitiesResponse.class);
 
         //testing the request
-        Future<Response> response = this.api.submitRequest(entitiesRequest);
+        ExecutorService threadPool = Executors.newFixedThreadPool(1);
+        Future<Response> response = threadPool.submit(entitiesRequest);
         assertInstanceOf(EntitiesResponse.class, response.get());
         assertEquals(response.get(), entitiesRequest.getResponse());
+        threadPool.shutdownNow();
     }
 
 
@@ -107,9 +108,11 @@ void errorResponse() throws ExecutionException, InterruptedException {
         RosetteRequest entitiesRequest = this.api.createRosetteRequest(AbstractRosetteAPI.ENTITIES_SERVICE_PATH, entitiesRequestData, EntitiesResponse.class);
 
         //testing the request
-        Future<Response> response = this.api.submitRequest(entitiesRequest);
+        ExecutorService threadPool = Executors.newFixedThreadPool(1);
+        Future<Response> response = threadPool.submit(entitiesRequest);
         assertInstanceOf(ErrorResponse.class, response.get());
         assertEquals(response.get(), entitiesRequest.getResponse());
+        threadPool.shutdownNow();
     }
 
     @Test
@@ -144,8 +147,9 @@ void testTiming() throws ExecutionException, InterruptedException {
         }
 
         //run requests
+        ExecutorService threadPool = Executors.newFixedThreadPool(7);
         Date d1 = new Date();
-        List<Future<Response>> responses = this.api.submitRequests(requests);
+        List<Future<Response>> responses = threadPool.invokeAll(requests);
         for (int i = 0; i < responses.size(); i++) {
             responses.get(i).get();
         }
@@ -168,7 +172,7 @@ void testTiming() throws ExecutionException, InterruptedException {
 
 
         d1 = new Date();
-        responses = this.api.submitRequests(requests);
+        responses = threadPool.invokeAll(requests);
         for (int i = 0; i < responses.size(); i++) {
             responses.get(i).get();
         }
@@ -178,54 +182,6 @@ void testTiming() throws ExecutionException, InterruptedException {
         assertTrue(d2.getTime() - d1.getTime() > requests.size() / concurrency * delay); // running faster than this would suggest it exceeds the maximum concurrency
     }
 
-    private RosetteRequest setupShutdownTest(int shutdownWaitSeconds, int responseDelayMillis, CloseableHttpClient client) {
-        this.api = new HttpRosetteAPI.Builder()
-                .url(String.format("http://localhost:%d/rest/v1", mockServer.getPort()))
-                .shutdownWait(shutdownWaitSeconds)
-                .httpClient(client)
-                .build();
-
-        //response setup
-        String entitiesResponse = "{\"entities\" : [ {     \"type\" : \"ORGANIZATION\",     \"mention\" : \"Securities and Exchange Commission\",     \"normalized\" : \"U.S. Securities and Exchange Commission\",     \"count\" : 1,     \"mentionOffsets\" : [ {       \"startOffset\" : 4,       \"endOffset\" : 38     } ],     \"entityId\" : \"Q953944\",     \"confidence\" : 0.39934742,     \"linkingConfidence\" : 0.67404154   } ] }";
-        setupResponse("/rest/v1/entities", entitiesResponse, 200, responseDelayMillis, 1);
-
-        //request setup
-        String entitiesTextData = "The Securities and Exchange Commission today announced the leadership of the agency’s trial unit.";
-        DocumentRequest<EntitiesOptions> entitiesRequestData = DocumentRequest.<EntitiesOptions>builder()
-                .content(entitiesTextData)
-                .build();
-        return this.api.createRosetteRequest(AbstractRosetteAPI.ENTITIES_SERVICE_PATH, entitiesRequestData, EntitiesResponse.class);
-    }
-
-    @Test
-    void successfulShutdown() throws IOException {
-        /* creating http client to avoid closing the httpClient of HttpRosetteApi during close()
-         *  which automatically closes the pending threads. This way the shutdown behaviour can be tested.*/
-        CloseableHttpClient httpClient = HttpClients.createDefault();
-        int shutdownWait = 3;
-        RosetteRequest request = setupShutdownTest(shutdownWait, 1000, httpClient);
-        this.api.submitRequest(request);
-        this.api.close();
-        httpClient.close();
-
-        assertInstanceOf(EntitiesResponse.class, request.getResponse()); //got response successfully
-    }
-
-    @Test
-    void timeoutShutdown() throws IOException {
-        /* creating http client to avoid closing the httpClient of HttpRosetteApi during close()
-         *  which automatically closes the pending threads. This way the shutdown behaviour can be tested.*/
-        CloseableHttpClient httpClient = HttpClients.createDefault();
-        int shutdownWait = 3;
-        RosetteRequest request = setupShutdownTest(shutdownWait, 5000, httpClient);
-        this.api.submitRequest(request);
-        this.api.close();
-
-        httpClient.close();
-        assertNull(request.getResponse()); //didn't get a response, it was forcefully shutdown
-    }
-
-
     @AfterEach
     void after() throws IOException {
         this.api.close();

examples/src/main/java/com/basistech/rosette/examples/ConcurrencyExample.java

@@ -37,6 +37,8 @@
 import java.util.Arrays;
 import java.util.List;
 import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
 import java.util.concurrent.Future;
 
 import static com.basistech.rosette.api.common.AbstractRosetteAPI.ENTITIES_SERVICE_PATH;
@@ -68,7 +70,7 @@ private  void run() throws IOException, ExecutionException, InterruptedException
                 .connectionConcurrency(maximumConcurrency)
                 .build();
 
-        List<RosetteRequest> threads = new ArrayList<>();
+        List<RosetteRequest> requests = new ArrayList<>();
         // Setting up entities request
         String entitiesTextData =
                 "The Securities and Exchange Commission today announced the leadership of the agency’s trial unit. "
@@ -77,21 +79,21 @@ private  void run() throws IOException, ExecutionException, InterruptedException
                 + "Since December 2016, Ms. Fitzpatrick and Mr. Gottesman have served as Co-Acting Chief Litigation Counsel.  "
                 + "In that role, they were jointly responsible for supervising the trial unit at the agency’s Washington D.C. headquarters "
                 + "as well as coordinating with litigators in the SEC’s 11 regional offices around the country.";
-        threads.add(
+        requests.add(
             rosetteApi.createRosetteRequest(ENTITIES_SERVICE_PATH,
                     DocumentRequest.<EntitiesOptions>builder().content(entitiesTextData).build(),
                     EntitiesResponse.class)
         );
         // Setting up language request
         String languageData = "Por favor Señorita, says the man.";
-        threads.add(
+        requests.add(
                 rosetteApi.createRosetteRequest(LANGUAGE_SERVICE_PATH,
                         DocumentRequest.<LanguageOptions>builder().content(languageData).build(),
                         LanguageResponse.class)
         );
         // Setting up morphology request
         // No content is given to this request and it will return an error response
-        threads.add(
+        requests.add(
                 rosetteApi.createRosetteRequest(MORPHOLOGY_SERVICE_PATH + "/" + MorphologicalFeature.COMPLETE,
                         DocumentRequest.<MorphologyOptions>builder().build(),
                         MorphologyResponse.class)
@@ -105,38 +107,39 @@ private  void run() throws IOException, ExecutionException, InterruptedException
             names.add(Name.builder().text(name).build());
         }
         double threshold = 0.75;
-        threads.add(
+        requests.add(
                 rosetteApi.createRosetteRequest(NAME_DEDUPLICATION_SERVICE_PATH,
                         NameDeduplicationRequest.builder().names(names).threshold(threshold).build(),
                         NameDeduplicationResponse.class)
         );
         //Setting up the tokens request
         String tokensData = "北京大学生物系主任办公室内部会议";
-        threads.add(
+        requests.add(
                 rosetteApi.createRosetteRequest(TOKENS_SERVICE_PATH,
                         DocumentRequest.builder().content(tokensData).build(),
                         TokensResponse.class)
         );
         //Setting up the transliteration request
         String transliterationData = "ana r2ye7 el gam3a el sa3a 3 el 3asr";
-        threads.add(
+        requests.add(
                 rosetteApi.createRosetteRequest(TRANSLITERATION_SERVICE_PATH,
                         DocumentRequest.builder().content(transliterationData).build(),
                         TransliterationResponse.class)
         );
 
         // start the threads
-        List<Future<Response>> futures = rosetteApi.submitRequests(threads);
+        ExecutorService threadPool = Executors.newFixedThreadPool(maximumConcurrency);
+        List<Future<Response>> futures = threadPool.invokeAll(requests);
 
         // wait for the threads to finish
-        for (int i = 0; i < threads.size(); i++) {
+        for (int i = 0; i < requests.size(); i++) {
             futures.get(i).get();
         }
 
-        for (int i = 0; i < threads.size(); i++) {
-            System.out.println(responseToJson(threads.get(i).getResponse()));
+        for (int i = 0; i < requests.size(); i++) {
+            System.out.println(responseToJson(requests.get(i).getResponse()));
         }
-        rosetteApi.close();
+        threadPool.shutdown();
     }