rschmitt commented on code in PR #782:
URL:
https://github.com/apache/httpcomponents-client/pull/782#discussion_r2691701945
##########
httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java:
##########
@@ -39,15 +48,93 @@
*/
public class Async {
+ private static final int DEFAULT_MAX_THREADS =
+ Math.max(2, Math.min(32,
Runtime.getRuntime().availableProcessors() * 2));
+
+ private static final int DEFAULT_QUEUE_CAPACITY = 1000;
+
+ private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);
+
private Executor executor;
private java.util.concurrent.Executor concurrentExec;
+ private ExecutorService ownedConcurrentExec;
Review Comment:
These fields need to either be `volatile`, `final`, or guarded by some other
memory barrier (e.g. only accessed from `synchronized` methods)
##########
httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java:
##########
@@ -122,4 +231,112 @@ public Future<Content> execute(final Request request) {
return execute(request, new ContentResponseHandler(), null);
}
+ /**
+ * Executes the given request asynchronously and returns a {@link
CompletableFuture} that completes
+ * when the response has been fully received and converted by the given
response handler.
+ *
+ * @param request the request to execute.
+ * @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) {
Review Comment:
Is there a reason why we can't push this down into `HttpAsyncClient`? Seems
like this would work as a simple `default` interface method to adapt the older
continuation-passing API to the new promise (`CompletableFuture`) based API.
##########
httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java:
##########
@@ -39,15 +48,93 @@
*/
public class Async {
+ private static final int DEFAULT_MAX_THREADS =
+ Math.max(2, Math.min(32,
Runtime.getRuntime().availableProcessors() * 2));
+
+ private static final int DEFAULT_QUEUE_CAPACITY = 1000;
+
+ private static final AtomicInteger INSTANCE_COUNT = new AtomicInteger(0);
+
private Executor executor;
private java.util.concurrent.Executor concurrentExec;
+ private ExecutorService ownedConcurrentExec;
+
+ private int maxThreads = DEFAULT_MAX_THREADS;
+ private int queueCapacity = DEFAULT_QUEUE_CAPACITY;
public static Async newInstance() {
return new Async();
}
Async() {
super();
+ // Keep legacy behavior by default.
+ }
+
+ public Async maxThreads(final int maxThreads) {
+ Args.positive(maxThreads, "maxThreads");
+ this.maxThreads = maxThreads;
+ rebuildOwnedExecutorIfActive();
+ return this;
+ }
+
+ public Async queueCapacity(final int queueCapacity) {
+ Args.positive(queueCapacity, "queueCapacity");
+ this.queueCapacity = queueCapacity;
+ rebuildOwnedExecutorIfActive();
+ return this;
+ }
+
+ /**
+ * Enables an owned bounded default executor for asynchronous request
execution.
+ *
+ * @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) {
+ Args.positive(maxThreads, "maxThreads");
+ Args.positive(queueCapacity, "queueCapacity");
+ this.maxThreads = maxThreads;
+ this.queueCapacity = queueCapacity;
+
+ shutdown();
+ this.ownedConcurrentExec = createDefaultExecutor(this.maxThreads,
this.queueCapacity);
+ this.concurrentExec = this.ownedConcurrentExec;
+ return this;
+ }
+
+ private void rebuildOwnedExecutorIfActive() {
+ if (this.ownedConcurrentExec != null) {
+ shutdown();
+ this.ownedConcurrentExec = createDefaultExecutor(this.maxThreads,
this.queueCapacity);
+ this.concurrentExec = this.ownedConcurrentExec;
+ }
+ }
+
+ 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;
+ };
Review Comment:
Use `DefaultThreadFactory` from core
##########
httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java:
##########
@@ -39,15 +48,93 @@
*/
public class Async {
Review Comment:
This class needs a thread-safety policy
##########
httpclient5-fluent/src/main/java/org/apache/hc/client5/http/fluent/Async.java:
##########
@@ -122,4 +231,112 @@ public Future<Content> execute(final Request request) {
return execute(request, new ContentResponseHandler(), null);
}
+ /**
+ * Executes the given request asynchronously and returns a {@link
CompletableFuture} that completes
+ * when the response has been fully received and converted by the given
response handler.
+ *
+ * @param request the request to execute.
+ * @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) {
Review Comment:
Wait, I'm confused. This class provides an async API for synchronous request
execution?
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: [email protected]
For queries about this service, please contact Infrastructure at:
[email protected]
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
