HTTPCLIENT-1307: Future based asynchronous request execution
Contributed by Jilles van Gurp <jilles at jillesvangurp.com> git-svn-id: https://svn.apache.org/repos/asf/httpcomponents/httpclient/trunk@1442152 13f79535-47bb-0310-9956-ffa450edef68
This commit is contained in:
Oleg Kalnichevski
parent
305357a488
commit
5cd571a159
|
|
@ -1,6 +1,9 @@
|
|||
Changes since 4.3 ALPHA1
|
||||
-------------------
|
||||
|
||||
* [HTTPCLIENT-1307] Future based asynchronous request execution.
|
||||
Contributed by Jilles van Gurp <jilles at jillesvangurp.com>
|
||||
|
||||
* [HTTPCLIENT-1311] Interrupt flag is not preserved where InterruptedException is caught.
|
||||
Contributed by Oleg Kalnichevski <olegk at apache.org>
|
||||
|
||||
|
|
|
|||
|
|
@ -0,0 +1,96 @@
|
|||
/*
|
||||
* ====================================================================
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
* ====================================================================
|
||||
*
|
||||
* This software consists of voluntary contributions made by many
|
||||
* individuals on behalf of the Apache Software Foundation. For more
|
||||
* information on the Apache Software Foundation, please see
|
||||
* <http://www.apache.org/>.
|
||||
*
|
||||
*/
|
||||
package org.apache.http.client.async;
|
||||
|
||||
import java.util.concurrent.atomic.AtomicLong;
|
||||
|
||||
/**
|
||||
* Collection of different counters used to gather metrics for {@link HttpAsyncClientWithFuture}.
|
||||
*/
|
||||
public class ConnectionMetrics {
|
||||
|
||||
final AtomicLong activeConnections = new AtomicLong();
|
||||
final AtomicLong scheduledConnections = new AtomicLong();
|
||||
final DurationCounter successfulConnections = new DurationCounter();
|
||||
final DurationCounter failedConnections = new DurationCounter();
|
||||
final DurationCounter requests = new DurationCounter();
|
||||
final DurationCounter tasks = new DurationCounter();
|
||||
|
||||
public ConnectionMetrics() {
|
||||
}
|
||||
|
||||
public String metricsAsJson() {
|
||||
final StringBuilder buf = new StringBuilder();
|
||||
buf.append("{\n");
|
||||
buf.append(" \"totalConnections\":" + requests.count() + ",\n");
|
||||
buf.append(" \"failedConnections\":" + failedConnections + ",\n");
|
||||
buf.append(" \"successfulConnections\":" + successfulConnections + ",\n");
|
||||
buf.append(" \"averageRequestDuration\":" + requests.averageDuration() + ",\n");
|
||||
buf.append(" \"averageTaskDuration\":" + tasks.averageDuration() + ",\n");
|
||||
buf.append(" \"activeConnections\":" + activeConnections + ",\n");
|
||||
buf.append(" \"scheduledConnections\":" + scheduledConnections + "\n");
|
||||
buf.append("}\n");
|
||||
|
||||
return buf.toString();
|
||||
}
|
||||
|
||||
public long activeConnections() {
|
||||
return activeConnections.get();
|
||||
}
|
||||
|
||||
public long scheduledConnections() {
|
||||
return scheduledConnections.get();
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return metricsAsJson();
|
||||
}
|
||||
|
||||
/**
|
||||
* A counter that can measure duration and number of events.
|
||||
*/
|
||||
public static class DurationCounter {
|
||||
private final AtomicLong count = new AtomicLong(0);
|
||||
private final AtomicLong cumulativeDuration = new AtomicLong(0);
|
||||
|
||||
public void increment(final long startTime) {
|
||||
count.incrementAndGet();
|
||||
cumulativeDuration.addAndGet(System.currentTimeMillis() - startTime);
|
||||
}
|
||||
|
||||
public long count() {
|
||||
return count.get();
|
||||
}
|
||||
|
||||
public long averageDuration() {
|
||||
final long counter = count.get();
|
||||
return cumulativeDuration.get() / counter;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,122 @@
|
|||
/*
|
||||
* ====================================================================
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
* ====================================================================
|
||||
*
|
||||
* This software consists of voluntary contributions made by many
|
||||
* individuals on behalf of the Apache Software Foundation. For more
|
||||
* information on the Apache Software Foundation, please see
|
||||
* <http://www.apache.org/>.
|
||||
*
|
||||
*/
|
||||
package org.apache.http.client.async;
|
||||
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import org.apache.http.client.HttpClient;
|
||||
import org.apache.http.client.ResponseHandler;
|
||||
import org.apache.http.client.methods.HttpUriRequest;
|
||||
import org.apache.http.concurrent.FutureCallback;
|
||||
import org.apache.http.protocol.HttpContext;
|
||||
|
||||
/**
|
||||
* Implementation of Callable that is wrapped with a {@link HttpAsyncClientFutureTask} by
|
||||
* {@link HttpAsyncClientWithFuture}. The callable orchestrates the invocation of
|
||||
* {@link HttpClient#execute(HttpUriRequest, ResponseHandler, HttpContext)} and callbacks in
|
||||
* {@link HttpAsyncClientCallback}.
|
||||
*
|
||||
* @param <V>
|
||||
* type returned by the responseHandler
|
||||
*/
|
||||
final class HttpAsyncClientCallable<V> implements Callable<V> {
|
||||
|
||||
private final HttpUriRequest request;
|
||||
|
||||
private final HttpClient httpclient;
|
||||
|
||||
private final AtomicBoolean cancelled = new AtomicBoolean(false);
|
||||
|
||||
final long scheduled = System.currentTimeMillis();
|
||||
long started = -1;
|
||||
long ended = -1;
|
||||
|
||||
private final HttpContext context;
|
||||
private final ResponseHandler<V> responseHandler;
|
||||
private final FutureCallback<V> callback;
|
||||
|
||||
private final ConnectionMetrics metrics;
|
||||
|
||||
HttpAsyncClientCallable(
|
||||
final HttpClient httpClient,
|
||||
final HttpUriRequest request,
|
||||
final HttpContext context,
|
||||
final ResponseHandler<V> responseHandler,
|
||||
final FutureCallback<V> callback,
|
||||
final ConnectionMetrics metrics) {
|
||||
this.httpclient = httpClient;
|
||||
this.responseHandler = responseHandler;
|
||||
this.request = request;
|
||||
this.context = context;
|
||||
this.callback = callback;
|
||||
this.metrics = metrics;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see java.util.concurrent.Callable#call()
|
||||
*/
|
||||
public V call() throws Exception {
|
||||
if (!cancelled.get()) {
|
||||
try {
|
||||
metrics.activeConnections.incrementAndGet();
|
||||
started = System.currentTimeMillis();
|
||||
try {
|
||||
metrics.scheduledConnections.decrementAndGet();
|
||||
final V result = httpclient.execute(request, responseHandler, context);
|
||||
ended = System.currentTimeMillis();
|
||||
metrics.successfulConnections.increment(started);
|
||||
if (callback != null) {
|
||||
callback.completed(result);
|
||||
}
|
||||
return result;
|
||||
} catch (final Exception e) {
|
||||
metrics.failedConnections.increment(started);
|
||||
ended = System.currentTimeMillis();
|
||||
if (callback != null) {
|
||||
callback.failed(e);
|
||||
}
|
||||
throw e;
|
||||
}
|
||||
} finally {
|
||||
metrics.requests.increment(started);
|
||||
metrics.tasks.increment(started);
|
||||
metrics.activeConnections.decrementAndGet();
|
||||
}
|
||||
} else {
|
||||
throw new IllegalStateException("call has been cancelled for request " + request.getURI());
|
||||
}
|
||||
}
|
||||
|
||||
public void cancel() {
|
||||
cancelled.set(true);
|
||||
if (callback != null) {
|
||||
callback.cancelled();
|
||||
}
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,118 @@
|
|||
/*
|
||||
* ====================================================================
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
* ====================================================================
|
||||
*
|
||||
* This software consists of voluntary contributions made by many
|
||||
* individuals on behalf of the Apache Software Foundation. For more
|
||||
* information on the Apache Software Foundation, please see
|
||||
* <http://www.apache.org/>.
|
||||
*
|
||||
*/
|
||||
package org.apache.http.client.async;
|
||||
|
||||
import java.util.concurrent.FutureTask;
|
||||
|
||||
import org.apache.http.client.methods.HttpUriRequest;
|
||||
|
||||
/**
|
||||
* FutureTask implementation that wraps a HttpAsyncClientCallable and exposes various task
|
||||
* specific metrics.
|
||||
*
|
||||
* @param <V>
|
||||
*/
|
||||
public class HttpAsyncClientFutureTask<V> extends FutureTask<V> {
|
||||
|
||||
private final HttpUriRequest request;
|
||||
|
||||
private final HttpAsyncClientCallable<V> callable;
|
||||
|
||||
public HttpAsyncClientFutureTask(
|
||||
final HttpUriRequest request,
|
||||
final HttpAsyncClientCallable<V> httpCallable) {
|
||||
super(httpCallable);
|
||||
this.request = request;
|
||||
this.callable = httpCallable;
|
||||
}
|
||||
|
||||
/*
|
||||
* (non-Javadoc)
|
||||
* @see java.util.concurrent.FutureTask#cancel(boolean)
|
||||
*/
|
||||
@Override
|
||||
public boolean cancel(final boolean mayInterruptIfRunning) {
|
||||
callable.cancel();
|
||||
if (mayInterruptIfRunning) {
|
||||
request.abort();
|
||||
}
|
||||
return super.cancel(mayInterruptIfRunning);
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in millis the task was scheduled.
|
||||
*/
|
||||
public long scheduledTime() {
|
||||
return callable.scheduled;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in millis the task was started.
|
||||
*/
|
||||
public long startedTime() {
|
||||
return callable.started;
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in millis the task was finished/cancelled.
|
||||
*/
|
||||
public long endedTime() {
|
||||
if (isDone()) {
|
||||
return callable.ended;
|
||||
} else {
|
||||
throw new IllegalStateException("Task is not done yet");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in millis it took to make the request (excluding the time it was
|
||||
* scheduled to be executed).
|
||||
*/
|
||||
public long requestDuration() {
|
||||
if (isDone()) {
|
||||
return endedTime() - startedTime();
|
||||
} else {
|
||||
throw new IllegalStateException("Task is not done yet");
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return the time in millis it took to execute the task from the moment it was scheduled.
|
||||
*/
|
||||
public long taskDuration() {
|
||||
if (isDone()) {
|
||||
return endedTime() - scheduledTime();
|
||||
} else {
|
||||
throw new IllegalStateException("Task is not done yet");
|
||||
}
|
||||
}
|
||||
|
||||
@Override
|
||||
public String toString() {
|
||||
return request.getRequestLine().getUri();
|
||||
}
|
||||
}
|
||||
|
|
@ -0,0 +1,194 @@
|
|||
/*
|
||||
* ====================================================================
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
* ====================================================================
|
||||
*
|
||||
* This software consists of voluntary contributions made by many
|
||||
* individuals on behalf of the Apache Software Foundation. For more
|
||||
* information on the Apache Software Foundation, please see
|
||||
* <http://www.apache.org/>.
|
||||
*
|
||||
*/
|
||||
package org.apache.http.client.async;
|
||||
|
||||
import java.util.ArrayList;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.Callable;
|
||||
import java.util.concurrent.ExecutorService;
|
||||
import java.util.concurrent.Executors;
|
||||
import java.util.concurrent.Future;
|
||||
import java.util.concurrent.TimeUnit;
|
||||
|
||||
import org.apache.http.annotation.ThreadSafe;
|
||||
import org.apache.http.client.HttpClient;
|
||||
import org.apache.http.client.ResponseHandler;
|
||||
import org.apache.http.client.methods.HttpUriRequest;
|
||||
import org.apache.http.concurrent.FutureCallback;
|
||||
import org.apache.http.protocol.HttpContext;
|
||||
|
||||
/**
|
||||
* HttpAsyncClientWithFuture wraps calls to execute with a {@link HttpAsyncClientFutureTask}
|
||||
* and schedules them using the provided executor service. Scheduled calls may be cancelled.
|
||||
* Similar to the non-blockcing HttpAsyncClient, a callback handler api is provided.
|
||||
*
|
||||
* @param <T>
|
||||
* the type of the value returned by the response handler you use
|
||||
*/
|
||||
@ThreadSafe
|
||||
public class HttpAsyncClientWithFuture {
|
||||
final HttpClient httpclient;
|
||||
|
||||
private final ExecutorService executorService;
|
||||
|
||||
private final ConnectionMetrics metrics = new ConnectionMetrics();
|
||||
|
||||
/**
|
||||
* Create a new HttpAsyncClientWithFuture.
|
||||
*
|
||||
* @param httpclient
|
||||
* you should tune your httpclient instance to match your needs. You should
|
||||
* align the max number of connections in the pool and the number of threads
|
||||
* in the executor; it doesn't make sense to have more threads than connections
|
||||
* and if you have less connections than threads, the threads will just end up
|
||||
* blocking on getting a connection from the pool.
|
||||
* @param executorService
|
||||
* any executorService will do here. E.g. {@link Executors#newFixedThreadPool(int)}
|
||||
*/
|
||||
public HttpAsyncClientWithFuture(
|
||||
final HttpClient httpclient,
|
||||
final ExecutorService executorService) {
|
||||
this.httpclient = httpclient;
|
||||
this.executorService = executorService;
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a request for execution.
|
||||
*
|
||||
* @param <T>
|
||||
*
|
||||
* @param request
|
||||
* request to execute
|
||||
* @param responseHandler
|
||||
* handler that will process the response.
|
||||
* @return HttpAsyncClientFutureTask for the scheduled request.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public <T> HttpAsyncClientFutureTask<T> execute(
|
||||
final HttpUriRequest request,
|
||||
final ResponseHandler<T> responseHandler) throws InterruptedException {
|
||||
return execute(request, null, responseHandler, null);
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule a request for execution.
|
||||
*
|
||||
* @param <T>
|
||||
*
|
||||
* @param request
|
||||
* request to execute
|
||||
* @param context
|
||||
* optional context; use null if not needed.
|
||||
* @param responseHandler
|
||||
* handler that will process the response.
|
||||
* @param callback
|
||||
* callback handler that will be called when the request is scheduled,
|
||||
* started, completed, failed, or cancelled.
|
||||
* @return HttpAsyncClientFutureTask for the scheduled request.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public <T> HttpAsyncClientFutureTask<T> execute(
|
||||
final HttpUriRequest request,
|
||||
final HttpContext context,
|
||||
final ResponseHandler<T> responseHandler,
|
||||
final FutureCallback<T> callback) throws InterruptedException {
|
||||
metrics.scheduledConnections.incrementAndGet();
|
||||
final HttpAsyncClientCallable<T> callable = new HttpAsyncClientCallable<T>(
|
||||
httpclient, request, context, responseHandler, callback, metrics);
|
||||
final HttpAsyncClientFutureTask<T> httpRequestFutureTask = new HttpAsyncClientFutureTask<T>(
|
||||
request, callable);
|
||||
executorService.execute(httpRequestFutureTask);
|
||||
|
||||
return httpRequestFutureTask;
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule multiple requests for execution.
|
||||
*
|
||||
* @param <T>
|
||||
*
|
||||
* @param responseHandler
|
||||
* handler that will process the responses.
|
||||
* @param requests
|
||||
* one or more requests.
|
||||
* @return a list of HttpAsyncClientFutureTask for the scheduled requests.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public <T> List<Future<T>> executeMultiple(
|
||||
final ResponseHandler<T> responseHandler,
|
||||
final HttpUriRequest... requests) throws InterruptedException {
|
||||
return executeMultiple(null, responseHandler, null, -1, null, requests);
|
||||
}
|
||||
|
||||
/**
|
||||
* Schedule multiple requests for execution with a timeout.
|
||||
*
|
||||
* @param <T>
|
||||
*
|
||||
* @param context
|
||||
* optional context; use null if not needed.
|
||||
* @param responseHandler
|
||||
* handler that will process the responses.
|
||||
* @param callback
|
||||
* callback handler that will be called when requests are scheduled,
|
||||
* started, completed, failed, or cancelled.
|
||||
* @param timeout
|
||||
* @param timeUnit
|
||||
* @param requests
|
||||
* one or more requests.
|
||||
* @return a list of HttpAsyncClientFutureTask for the scheduled requests.
|
||||
* @throws InterruptedException
|
||||
*/
|
||||
public <T> List<Future<T>> executeMultiple(
|
||||
final HttpContext context,
|
||||
final ResponseHandler<T> responseHandler,
|
||||
final FutureCallback<T> callback,
|
||||
final long timeout, final TimeUnit timeUnit,
|
||||
final HttpUriRequest... requests) throws InterruptedException {
|
||||
metrics.scheduledConnections.incrementAndGet();
|
||||
final List<Callable<T>> callables = new ArrayList<Callable<T>>();
|
||||
for (final HttpUriRequest request : requests) {
|
||||
final HttpAsyncClientCallable<T> callable = new HttpAsyncClientCallable<T>(
|
||||
httpclient, request, context, responseHandler, callback, metrics);
|
||||
callables.add(callable);
|
||||
}
|
||||
if (timeout > 0) {
|
||||
return executorService.invokeAll(callables, timeout, timeUnit);
|
||||
} else {
|
||||
return executorService.invokeAll(callables);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* @return metrics gathered for this instance.
|
||||
* @see ConnectionMetrics.
|
||||
*/
|
||||
public ConnectionMetrics metrics() {
|
||||
return metrics;
|
||||
}
|
||||
|
||||
}
|
||||
|
|
@ -0,0 +1,180 @@
|
|||
/*
|
||||
* ====================================================================
|
||||
* Licensed to the Apache Software Foundation (ASF) under one
|
||||
* or more contributor license agreements. See the NOTICE file
|
||||
* distributed with this work for additional information
|
||||
* regarding copyright ownership. The ASF licenses this file
|
||||
* to you under the Apache License, Version 2.0 (the
|
||||
* "License"); you may not use this file except in compliance
|
||||
* with the License. You may obtain a copy of the License at
|
||||
*
|
||||
* http://www.apache.org/licenses/LICENSE-2.0
|
||||
*
|
||||
* Unless required by applicable law or agreed to in writing,
|
||||
* software distributed under the License is distributed on an
|
||||
* "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
|
||||
* KIND, either express or implied. See the License for the
|
||||
* specific language governing permissions and limitations
|
||||
* under the License.
|
||||
* ====================================================================
|
||||
*
|
||||
* This software consists of voluntary contributions made by many
|
||||
* individuals on behalf of the Apache Software Foundation. For more
|
||||
* information on the Apache Software Foundation, please see
|
||||
* <http://www.apache.org/>.
|
||||
*
|
||||
*/
|
||||
package org.apache.http.client.async;
|
||||
|
||||
import java.io.IOException;
|
||||
import java.net.InetSocketAddress;
|
||||
import java.util.List;
|
||||
import java.util.concurrent.CancellationException;
|
||||
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 java.util.concurrent.TimeoutException;
|
||||
import java.util.concurrent.atomic.AtomicBoolean;
|
||||
|
||||
import junit.framework.Assert;
|
||||
|
||||
import org.apache.http.HttpException;
|
||||
import org.apache.http.HttpRequest;
|
||||
import org.apache.http.HttpResponse;
|
||||
import org.apache.http.client.ClientProtocolException;
|
||||
import org.apache.http.client.ResponseHandler;
|
||||
import org.apache.http.client.methods.HttpGet;
|
||||
import org.apache.http.concurrent.FutureCallback;
|
||||
import org.apache.http.impl.client.CloseableHttpClient;
|
||||
import org.apache.http.impl.client.HttpClientBuilder;
|
||||
import org.apache.http.localserver.LocalTestServer;
|
||||
import org.apache.http.protocol.HttpContext;
|
||||
import org.apache.http.protocol.HttpRequestHandler;
|
||||
import org.junit.After;
|
||||
import org.junit.Before;
|
||||
import org.junit.Test;
|
||||
|
||||
public class HttpClientWithFutureTest {
|
||||
|
||||
private LocalTestServer localServer;
|
||||
private String uri;
|
||||
private HttpAsyncClientWithFuture httpAsyncClientWithFuture;
|
||||
private CloseableHttpClient httpClient;
|
||||
private ExecutorService executorService;
|
||||
|
||||
private final AtomicBoolean blocked = new AtomicBoolean(false);
|
||||
|
||||
@Before
|
||||
public void before() throws Exception {
|
||||
this.localServer = new LocalTestServer(null, null);
|
||||
this.localServer.register("/wait", new HttpRequestHandler() {
|
||||
|
||||
public void handle(
|
||||
final HttpRequest request, final HttpResponse response,
|
||||
final HttpContext context) throws HttpException, IOException {
|
||||
try {
|
||||
while(blocked.get()) {
|
||||
Thread.sleep(10);
|
||||
}
|
||||
} catch (final InterruptedException e) {
|
||||
throw new IllegalStateException(e);
|
||||
}
|
||||
response.setStatusCode(200);
|
||||
}
|
||||
});
|
||||
this.localServer.start();
|
||||
final InetSocketAddress address = localServer.getServiceAddress();
|
||||
uri = "http://" + address.getHostName() + ":" + address.getPort() + "/wait";
|
||||
|
||||
httpClient = HttpClientBuilder.create().setMaxConnPerRoute(5).setMaxConnTotal(5).build();
|
||||
executorService = Executors.newFixedThreadPool(5);
|
||||
httpAsyncClientWithFuture = new HttpAsyncClientWithFuture(httpClient, executorService);
|
||||
}
|
||||
|
||||
@After
|
||||
public void after() throws Exception {
|
||||
blocked.set(false); // any remaining requests should unblock
|
||||
this.localServer.stop();
|
||||
httpClient.close();
|
||||
executorService.shutdownNow();
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldExecuteSingleCall() throws InterruptedException, ExecutionException {
|
||||
final HttpAsyncClientFutureTask<Boolean> task = httpAsyncClientWithFuture.execute(
|
||||
new HttpGet(uri), new OkidokiHandler());
|
||||
Assert.assertTrue("request should have returned OK", task.get());
|
||||
}
|
||||
|
||||
@Test(expected=CancellationException.class)
|
||||
public void shouldCancel() throws InterruptedException, ExecutionException {
|
||||
final HttpAsyncClientFutureTask<Boolean> task = httpAsyncClientWithFuture.execute(
|
||||
new HttpGet(uri), new OkidokiHandler());
|
||||
task.cancel(true);
|
||||
task.get();
|
||||
}
|
||||
|
||||
@Test(expected=TimeoutException.class)
|
||||
public void shouldTimeout() throws InterruptedException, ExecutionException, TimeoutException {
|
||||
blocked.set(true);
|
||||
final HttpAsyncClientFutureTask<Boolean> task = httpAsyncClientWithFuture.execute(
|
||||
new HttpGet(uri), new OkidokiHandler());
|
||||
task.get(10, TimeUnit.MILLISECONDS);
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldExecuteMultipleCalls() throws InterruptedException, ExecutionException {
|
||||
final HttpGet[] requests= new HttpGet[100];
|
||||
for(int i=0;i<100;i++) {
|
||||
requests[i]=new HttpGet(uri);
|
||||
}
|
||||
final List<Future<Boolean>> tasks = httpAsyncClientWithFuture.executeMultiple(
|
||||
new OkidokiHandler(), requests);
|
||||
for (final Future<Boolean> task : tasks) {
|
||||
Assert.assertTrue("request should have returned OK", task.get());
|
||||
}
|
||||
}
|
||||
|
||||
@Test
|
||||
public void shouldExecuteMultipleCallsAndCallback() throws InterruptedException, ExecutionException {
|
||||
final HttpGet[] requests= new HttpGet[100];
|
||||
for(int i=0;i<100;i++) {
|
||||
requests[i]=new HttpGet(uri);
|
||||
}
|
||||
final CountingCallback callback = new CountingCallback();
|
||||
httpAsyncClientWithFuture.executeMultiple(null,
|
||||
new OkidokiHandler(), callback , 1000, TimeUnit.MILLISECONDS, requests);
|
||||
Assert.assertEquals(100, callback.completed);
|
||||
Assert.assertEquals(0, callback.cancelled);
|
||||
Assert.assertEquals(0, callback.failed);
|
||||
}
|
||||
|
||||
|
||||
private final class CountingCallback implements FutureCallback<Boolean> {
|
||||
int failed=0;
|
||||
int cancelled=0;
|
||||
int completed=0;
|
||||
|
||||
public void failed(final Exception ex) {
|
||||
failed++;
|
||||
}
|
||||
|
||||
public void completed(final Boolean result) {
|
||||
completed++;
|
||||
}
|
||||
|
||||
public void cancelled() {
|
||||
cancelled++;
|
||||
}
|
||||
}
|
||||
|
||||
|
||||
private final class OkidokiHandler implements ResponseHandler<Boolean> {
|
||||
public Boolean handleResponse(final HttpResponse response) throws ClientProtocolException, IOException {
|
||||
return response.getStatusLine().getStatusCode() == 200;
|
||||
}
|
||||
}
|
||||
|
||||
}
|
||||
Loading…
Reference in New Issue