Added facility to execute requests asynchronously using fluent HC API

git-svn-id: https://svn.apache.org/repos/asf/httpcomponents/httpclient/trunk@1239732 13f79535-47bb-0310-9956-ffa450edef68

fluent-hc/src/examples/org/apache/http/client/fluent/FluentAsync.java

@@ -0,0 +1,88 @@
+/*
+ * ====================================================================
+ *
+ *  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.fluent;
+
+import java.util.LinkedList;
+import java.util.Queue;
+import java.util.concurrent.ExecutionException;
+import java.util.concurrent.ExecutorService;
+import java.util.concurrent.Executors;
+import java.util.concurrent.Future;
+
+import org.apache.http.concurrent.FutureCallback;
+
+/**
+ * This example demonstrates how the he HttpClient fluent API can be used to execute multiple
+ * requests asynchronously using background threads.
+ */
+public class FluentAsync {
+
+    public static void main(String[] args)throws Exception {
+        // Use pool of two threads
+        ExecutorService threadpool = Executors.newFixedThreadPool(2);
+        Async async = Async.newInstance().use(threadpool);
+        
+        Request[] requests = new Request[] {
+                Request.Get("http://www.google.com/"),
+                Request.Get("http://www.yahoo.com/"),
+                Request.Get("http://www.apache.com/"),
+                Request.Get("http://www.apple.com/")
+        };
+
+        
+        Queue<Future<Content>> queue = new LinkedList<Future<Content>>();
+        // Execute requests asynchronously
+        for (final Request request: requests) {
+            Future<Content> future = async.execute(request, new FutureCallback<Content>() {
+                
+                public void failed(final Exception ex) {
+                    System.out.println(ex.getMessage() + ": " + request);
+                }
+                
+                public void completed(final Content content) {
+                    System.out.println("Request completed: " + request);
+                }
+                
+                public void cancelled() {
+                }
+                
+            });
+            queue.add(future);
+        }
+
+        while(!queue.isEmpty()) {
+            Future<Content> future = queue.remove();
+            try {
+                future.get();
+            } catch (ExecutionException ex) {
+            }
+        }
+        System.out.println("Done");
+        threadpool.shutdown();
+    }
+
+}

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

@@ -0,0 +1,119 @@
+/*
+ * ====================================================================
+ *
+ *  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.fluent;
+
+import java.util.concurrent.Future;
+
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.concurrent.BasicFuture;
+import org.apache.http.concurrent.FutureCallback;
+
+public class Async {
+
+    private Executor executor;
+    private java.util.concurrent.Executor concurrentExec;
+
+    public static Async newInstance() {
+        return new Async();
+    }
+
+    Async() {
+        super();
+    }
+
+    public Async use(final Executor executor) {
+        this.executor = executor;
+        return this;
+    }
+    
+    public Async use(final java.util.concurrent.Executor concurrentExec) {
+        this.concurrentExec = concurrentExec;
+        return this;
+    }
+    
+    static class ExecRunnable<T> implements Runnable {
+        
+        private final BasicFuture<T> future;
+        private final Request request;
+        private final Executor executor;
+        private final ResponseHandler<T> handler;
+        
+        ExecRunnable(
+                final BasicFuture<T> future, 
+                final Request request, 
+                final Executor executor, 
+                final ResponseHandler<T> handler) {
+            super();
+            this.future = future;
+            this.request = request;
+            this.executor = executor;
+            this.handler = handler;
+        }
+
+        public void run() {
+            try {
+                Response response = this.executor.execute(this.request);
+                T result = response.handleResponse(this.handler);
+                this.future.completed(result);
+            } catch (Exception ex) {
+                this.future.failed(ex);
+            }
+        }
+        
+    }
+    
+    public <T> Future<T> execute(
+            final Request request, final ResponseHandler<T> handler, final FutureCallback<T> callback) {
+        BasicFuture<T> future = new BasicFuture<T>(callback);
+        ExecRunnable<T> runnable = new ExecRunnable<T>(
+                future,
+                request,
+                this.executor != null ? this.executor : Executor.newInstance(),
+                handler);
+        if (this.concurrentExec != null) {
+            this.concurrentExec.execute(runnable);
+        } else {
+            Thread t = new Thread(runnable);
+            t.setDaemon(true);
+            t.start();
+        }
+        return future;
+    }
+    
+    public <T> Future<T> execute(final Request request, final ResponseHandler<T> handler) {
+        return execute(request, handler, null);
+    }
+
+    public Future<Content> execute(final Request request, final FutureCallback<Content> callback) {
+        return execute(request, new ContentResponseHandler(), callback);
+    }
+    
+    public Future<Content> execute(final Request request) {
+        return execute(request, new ContentResponseHandler(), null);
+    }
+    
+}

fluent-hc/src/main/java/org/apache/http/client/fluent/Content.java

@@ -70,4 +70,9 @@ public class Content {
         return new ByteArrayInputStream(this.raw);
     }
 
+    @Override
+    public String toString() {
+        return asString();
+    }
+    
 }

fluent-hc/src/main/java/org/apache/http/client/fluent/ContentResponseHandler.java

@@ -0,0 +1,59 @@
+/*
+ * ====================================================================
+ *
+ *  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.fluent;
+
+import java.io.IOException;
+
+import org.apache.http.HttpEntity;
+import org.apache.http.HttpResponse;
+import org.apache.http.StatusLine;
+import org.apache.http.client.ClientProtocolException;
+import org.apache.http.client.HttpResponseException;
+import org.apache.http.client.ResponseHandler;
+import org.apache.http.entity.ContentType;
+import org.apache.http.util.EntityUtils;
+
+class ContentResponseHandler implements ResponseHandler<Content> {
+
+    public Content handleResponse(
+            final HttpResponse response) throws ClientProtocolException, IOException {
+        StatusLine statusLine = response.getStatusLine();
+        HttpEntity entity = response.getEntity();
+        if (statusLine.getStatusCode() >= 300) {
+            throw new HttpResponseException(statusLine.getStatusCode(),
+                    statusLine.getReasonPhrase());
+        }
+        if (entity != null) {
+            return new Content(
+                    EntityUtils.toByteArray(entity),
+                    ContentType.getOrDefault(entity));
+        } else {
+            return Content.NO_CONTENT;
+        }
+    }
+
+}

fluent-hc/src/main/java/org/apache/http/client/fluent/Executor.java

@@ -41,7 +41,6 @@ import org.apache.http.client.CredentialsProvider;
 import org.apache.http.client.HttpClient;
 import org.apache.http.client.methods.HttpRequestBase;
 import org.apache.http.client.protocol.ClientContext;
-import org.apache.http.conn.routing.HttpRoute;
 import org.apache.http.conn.scheme.Scheme;
 import org.apache.http.impl.auth.BasicScheme;
 import org.apache.http.impl.client.BasicAuthCache;
@@ -53,9 +52,16 @@ import org.apache.http.protocol.BasicHttpContext;
 
 public class Executor {
 
-    final static PoolingClientConnectionManager CONNMGR = new PoolingClientConnectionManager(
+    final static PoolingClientConnectionManager CONNMGR;
+    final static DefaultHttpClient CLIENT;
+    
+    static {
+        CONNMGR = new PoolingClientConnectionManager(
                 SchemeRegistryFactory.createSystemDefault());
-    final static DefaultHttpClient CLIENT = new DefaultHttpClient(CONNMGR);
+        CONNMGR.setDefaultMaxPerRoute(100);
+        CONNMGR.setMaxTotal(200);
+        CLIENT = new DefaultHttpClient(CONNMGR);
+    }
 
     public static Executor newInstance() {
         return new Executor(CLIENT);
@@ -142,27 +148,15 @@ public class Executor {
     }
 
     public Response execute(
-            final Request req) throws ClientProtocolException, IOException {
+            final Request request) throws ClientProtocolException, IOException {
         this.localContext.setAttribute(ClientContext.CREDS_PROVIDER, this.credentialsProvider);
         this.localContext.setAttribute(ClientContext.AUTH_CACHE, this.authCache);
         this.localContext.setAttribute(ClientContext.COOKIE_STORE, this.cookieStore);
-        HttpRequestBase httprequest = req.getHttpRequest();
+        HttpRequestBase httprequest = request.getHttpRequest();
         httprequest.reset();
         return new Response(this.httpclient.execute(httprequest, this.localContext));
     }
 
-    public static void setMaxTotal(int max) {
-        CONNMGR.setMaxTotal(max);
-    }
-
-    public static void setDefaultMaxPerRoute(int max) {
-        CONNMGR.setDefaultMaxPerRoute(max);
-    }
-
-    public static void setMaxPerRoute(final HttpRoute route, int max) {
-        CONNMGR.setMaxPerRoute(route, max);
-    }
-
     public static void registerScheme(final Scheme scheme) {
         CONNMGR.getSchemeRegistry().register(scheme);
     }

fluent-hc/src/main/java/org/apache/http/client/fluent/Request.java

@@ -306,4 +306,9 @@ public class Request {
         return body(new InputStreamEntity(instream, -1, contentType));
     }
 
+    @Override
+    public String toString() {
+        return this.request.getRequestLine().toString();
+    }
+    
 }

fluent-hc/src/main/java/org/apache/http/client/fluent/Response.java

@@ -83,26 +83,7 @@ public class Response {
     }
 
     public Content returnContent() throws ClientProtocolException, IOException {
-        return handleResponse(new ResponseHandler<Content>() {
+        return handleResponse(new ContentResponseHandler());
-
-            public Content handleResponse(
-                    final HttpResponse response) throws ClientProtocolException, IOException {
-                StatusLine statusLine = response.getStatusLine();
-                HttpEntity entity = response.getEntity();
-                if (statusLine.getStatusCode() >= 300) {
-                    throw new HttpResponseException(statusLine.getStatusCode(),
-                            statusLine.getReasonPhrase());
-                }
-                if (entity != null) {
-                    return new Content(
-                            EntityUtils.toByteArray(entity),
-                            ContentType.getOrDefault(entity));
-                } else {
-                    return Content.NO_CONTENT;
-                }
-            }
-
-        });
     }
 
     public HttpResponse returnResponse() throws IOException {