small changes

Author: arturobernalg

httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java

@@ -31,12 +31,14 @@
 import java.util.concurrent.Future;
 import java.util.concurrent.LinkedBlockingQueue;
 import java.util.concurrent.RejectedExecutionException;
-import java.util.concurrent.ThreadFactory;
 import java.util.concurrent.ThreadPoolExecutor;
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.atomic.AtomicInteger;
 
+import org.apache.hc.core5.annotation.Contract;
+import org.apache.hc.core5.annotation.ThreadingBehavior;
 import org.apache.hc.core5.concurrent.BasicFuture;
+import org.apache.hc.core5.concurrent.DefaultThreadFactory;
 import org.apache.hc.core5.concurrent.FutureCallback;
 import org.apache.hc.core5.http.io.HttpClientResponseHandler;
 import org.apache.hc.core5.util.Args;
@@ -46,6 +48,7 @@
  *
  * @since 4.3
  */
+@Contract(threading = ThreadingBehavior.SAFE_CONDITIONAL)
 public class Async {
 
     private static final int DEFAULT_MAX_THREADS =
@@ -56,8 +59,8 @@ public class Async {
     private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);
 
     private Executor executor;
-    private java.util.concurrent.Executor concurrentExec;
-    private ExecutorService ownedConcurrentExec;
+    private volatile java.util.concurrent.Executor concurrentExec;
+    private volatile ExecutorService ownedConcurrentExec;
 
     private int maxThreads = DEFAULT_MAX_THREADS;
     private int queueCapacity = DEFAULT_QUEUE_CAPACITY;
@@ -85,13 +88,23 @@ public Async queueCapacity(final int queueCapacity) {
         return this;
     }
 
+    /**
+     * Enables an owned bounded default executor for asynchronous request execution using the
+     * current {@code maxThreads} and {@code queueCapacity} settings.
+     *
+     * @return this instance.
+     * @since 5.7
+     */
+    public Async useDefaultExecutor() {
+        return useDefaultExecutor(this.maxThreads, this.queueCapacity);
+    }
+
     /**
      * Enables an owned bounded default executor for asynchronous request execution.
      *
-     * @param maxThreads maximum number of threads.
+     * @param maxThreads    maximum number of threads.
      * @param queueCapacity maximum number of queued tasks.
      * @return this instance.
-     *
      * @since 5.7
      */
     public Async useDefaultExecutor(final int maxThreads, final int queueCapacity) {
@@ -116,13 +129,9 @@ private void rebuildOwnedExecutorIfActive() {
 
     private static ExecutorService createDefaultExecutor(final int maxThreads, final int queueCapacity) {
         final int instanceId = INSTANCE_COUNT.incrementAndGet();
-        final AtomicInteger threadCount = new AtomicInteger(0);
-
-        final ThreadFactory threadFactory = r -> {
-            final Thread t = new Thread(r, "httpclient5-fluent-async-" + instanceId + "-" + threadCount.incrementAndGet());
-            t.setDaemon(true);
-            return t;
-        };
+        final DefaultThreadFactory threadFactory = new DefaultThreadFactory(
+                "httpclient5-fluent-async-" + instanceId + "-",
+                true);
 
         final ThreadPoolExecutor exec = new ThreadPoolExecutor(
                 maxThreads,
@@ -239,7 +248,6 @@ public Future<Content> execute(final Request request) {
      * @param handler the response handler.
      * @param <T>     the handler result type.
      * @return a {@code CompletableFuture} producing the handler result.
-     *
      * @since 5.7
      */
     public <T> CompletableFuture<T> executeAsync(final Request request, final HttpClientResponseHandler<T> handler) {
@@ -275,7 +283,6 @@ public void cancelled() {
      * @param callback the callback to invoke on completion, failure, or cancellation; may be {@code null}.
      * @param <T>      the handler result type.
      * @return a {@code CompletableFuture} producing the handler result.
-     *
      * @since 5.7
      */
     public <T> CompletableFuture<T> executeAsync(
@@ -317,7 +324,6 @@ public void cancelled() {
      *
      * @param request the request to execute.
      * @return a {@code CompletableFuture} producing the response {@code Content}.
-     *
      * @since 5.7
      */
     public CompletableFuture<Content> executeAsync(final Request request) {
@@ -332,7 +338,6 @@ public CompletableFuture<Content> executeAsync(final Request request) {
      * @param request  the request to execute.
      * @param callback the callback to invoke on completion, failure, or cancellation; may be {@code null}.
      * @return a {@code CompletableFuture} producing the response {@code Content}.
-     *
      * @since 5.7
      */
     public CompletableFuture<Content> executeAsync(final Request request, final FutureCallback<Content> callback) {

httpclient5-fluent/src/test/java/org/apache/hc/client5/http/examples/fluent/FluentAsyncCompletableFutureCallback.java

@@ -50,7 +50,7 @@ public static void main(final String... args) throws Exception {
                 Request.get("http://www.apple.com/")
         );
 
-        final Async async = Async.newInstance().useDefaultExecutor(8, 500);
+        final Async async = Async.newInstance().maxThreads(8).queueCapacity(500).useDefaultExecutor();
         try {
 
             final CompletableFuture<?>[] futures = requests.stream()