Add ListenableFuture support

jfarcand · jfarcand · commit 5ad43e2376eb · 2011-02-18T08:51:36.000-05:00
diff --git a/src/main/java/com/ning/http/client/ListenableFuture.java b/src/main/java/com/ning/http/client/ListenableFuture.java
@@ -0,0 +1,100 @@
+/*
+ * Copyright 2010 Ning, Inc.
+ *
+ * Ning 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.
+ */
+/*
+ * Copyright (C) 2007 Google Inc.
+ *
+ * Licensed 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.
+ */
+package com.ning.http.client;
+
+import java.util.concurrent.Callable;
+import java.util.concurrent.Executor;
+import java.util.concurrent.Future;
+
+/**
+ * Extended {@link Future}
+ *
+ * @param <V> Type of the value that will be returned.
+ */
+public interface ListenableFuture<V> extends Future<V> {
+
+    /**
+     * Execute a {@link Callable}  and if there is no exception, mark this Future as done and release the internal lock.
+     *
+     * @param callable
+     */
+    void done(Callable callable);
+
+    /**
+     * Abort the current processing, and propagate the {@link Throwable} to the {@link AsyncHandler} or {@link Future}
+     *
+     * @param t
+     */
+    void abort(Throwable t);
+
+    /**
+     * Set the content that will be returned by this instance
+     *
+     * @param v the content that will be returned by this instance
+     */
+    void content(V v);
+
+    /**
+     * Touch the current instance to prevent external service to times out.
+     */
+    void touch();
+
+    /**
+     * Write the {@link Request} headers
+     */
+    boolean getAndSetWriteHeaders(boolean writeHeader);
+
+    /**
+     * Write the {@link Request} body
+     */
+    boolean getAndSetWriteBody(boolean writeBody);
+
+    /**
+     * <p>Adds a listener and executor to the ListenableFuture.
+     * The listener will be {@linkplain java.util.concurrent.Executor#execute(Runnable) passed
+     * to the executor} for execution when the {@code Future}'s computation is
+     * {@linkplain Future#isDone() complete}.
+     * <p/>
+     * <p>There is no guaranteed ordering of execution of listeners, they may get
+     * called in the order they were added and they may get called out of order,
+     * but any listener added through this method is guaranteed to be called once
+     * the computation is complete.
+     *
+     * @param listener the listener to run when the computation is complete.
+     * @param exec     the executor to run the listener in.
+     * @return this Future
+     * @throws NullPointerException if the executor or listener was null.
+     * @throws java.util.concurrent.RejectedExecutionException
+     *                              if we tried to execute the listener
+     *                              immediately but the executor rejected it.
+     */
+    ListenableFuture<V> addListener(Runnable listener, Executor exec);
+}
diff --git a/src/main/java/com/ning/http/client/listenable/AbstractListenableFuture.java b/src/main/java/com/ning/http/client/listenable/AbstractListenableFuture.java
@@ -0,0 +1,70 @@
+/*
+ * Copyright (c) 2010-2011 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+/*
+ * Copyright (C) 2007 Google Inc.
+ *
+ * Licensed 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.
+ */
+
+package com.ning.http.client.listenable;
+
+import com.ning.http.client.ListenableFuture;
+
+import java.util.concurrent.Executor;
+
+/**
+ * <p>An abstract base implementation of the listener support provided by
+ * {@link ListenableFuture}. This class uses an {@link ExecutionList} to
+ * guarantee that all registered listeners will be executed. Listener/Executor
+ * pairs are stored in the execution list and executed in the order in which
+ * they were added, but because of thread scheduling issues there is no
+ * guarantee that the JVM will execute them in order. In addition, listeners
+ * added after the task is complete will be executed immediately, even if some
+ * previously added listeners have not yet been executed.
+ *
+ * @author Sven Mawson
+ * @since 1
+ */
+public abstract class AbstractListenableFuture<V> implements ListenableFuture<V> {
+
+    // The execution list to hold our executors.
+    private final ExecutionList executionList = new ExecutionList();
+
+    /*
+    * Adds a listener/executor pair to execution list to execute when this task
+    * is completed.
+    */
+
+    public ListenableFuture<V> addListener(Runnable listener, Executor exec) {
+        executionList.add(listener, exec);
+        return this;
+    }
+
+    /*
+    * Override the done method to execute the execution list.
+    */
+
+    protected void done() {
+        executionList.run();
+    }
+}
diff --git a/src/main/java/com/ning/http/client/listenable/ExecutionList.java b/src/main/java/com/ning/http/client/listenable/ExecutionList.java
@@ -0,0 +1,138 @@
+/*
+ * Copyright (c) 2010-2011 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+/*
+ * Copyright (C) 2007 Google Inc.
+ *
+ * Licensed 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.
+ */
+
+package com.ning.http.client.listenable;
+
+import java.util.Queue;
+import java.util.concurrent.Executor;
+import java.util.concurrent.LinkedBlockingQueue;
+import java.util.logging.Level;
+import java.util.logging.Logger;
+
+/**
+ * <p>A list of ({@code Runnable}, {@code Executor}) pairs that guarantees
+ * that every {@code Runnable} that is added using the add method will be
+ * executed in its associated {@code Executor} after {@link #run()} is called.
+ * {@code Runnable}s added after {@code run} is called are still guaranteed to
+ * execute.
+ *
+ * @author Nishant Thakkar
+ * @author Sven Mawson
+ * @since 1
+ */
+public final class ExecutionList implements Runnable {
+
+    // Logger to log exceptions caught when running runnables.
+    private static final Logger log =
+            Logger.getLogger(ExecutionList.class.getName());
+
+    // The runnable,executor pairs to execute.
+    private final Queue<RunnableExecutorPair> runnables = new LinkedBlockingQueue<RunnableExecutorPair>();
+
+    // Boolean we use mark when execution has started.  Only accessed from within
+    // synchronized blocks.
+    private boolean executed = false;
+
+    /**
+     * Add the runnable/executor pair to the list of pairs to execute.  Executes
+     * the pair immediately if we've already started execution.
+     */
+    public void add(Runnable runnable, Executor executor) {
+
+        if (runnable == null) {
+            throw new NullPointerException("Runnable is null");
+        }
+
+        if (executor == null) {
+            throw new NullPointerException("Executor is null");
+        }
+
+        boolean executeImmediate = false;
+
+        // Lock while we check state.  We must maintain the lock while adding the
+        // new pair so that another thread can't run the list out from under us.
+        // We only add to the list if we have not yet started execution.
+        synchronized (runnables) {
+            if (!executed) {
+                runnables.add(new RunnableExecutorPair(runnable, executor));
+            } else {
+                executeImmediate = true;
+            }
+        }
+
+        // Execute the runnable immediately.  Because of scheduling this may end up
+        // getting called before some of the previously added runnables, but we're
+        // ok with that.  If we want to change the contract to guarantee ordering
+        // among runnables we'd have to modify the logic here to allow it.
+        if (executeImmediate) {
+            executor.execute(runnable);
+        }
+    }
+
+    /**
+     * Runs this execution list, executing all pairs in the order they were
+     * added.  Pairs added after this method has started executing the list will
+     * be executed immediately.
+     */
+    public void run() {
+
+        // Lock while we update our state so the add method above will finish adding
+        // any listeners before we start to run them.
+        synchronized (runnables) {
+            executed = true;
+        }
+
+        // At this point the runnables will never be modified by another
+        // thread, so we are safe using it outside of the synchronized block.
+        while (!runnables.isEmpty()) {
+            runnables.poll().execute();
+        }
+    }
+
+    private static class RunnableExecutorPair {
+        final Runnable runnable;
+        final Executor executor;
+
+        RunnableExecutorPair(Runnable runnable, Executor executor) {
+            this.runnable = runnable;
+            this.executor = executor;
+        }
+
+        void execute() {
+            try {
+                executor.execute(runnable);
+            } catch (RuntimeException e) {
+                // Log it and keep going, bad runnable and/or executor.  Don't
+                // punish the other runnables if we're given a bad one.  We only
+                // catch RuntimeException because we want Errors to propagate up.
+                log.log(Level.SEVERE, "RuntimeException while executing runnable "
+                        + runnable + " with executor " + executor, e);
+            }
+        }
+    }
+}
diff --git a/src/test/java/com/ning/http/client/async/ListenableFutureTest.java b/src/test/java/com/ning/http/client/async/ListenableFutureTest.java
@@ -0,0 +1,54 @@
+/*
+ * Copyright 2010 Ning, Inc.
+ *
+ * Ning 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.
+ */
+package com.ning.http.client.async;
+
+import com.ning.http.client.AsyncHttpClient;
+import com.ning.http.client.ListenableFuture;
+import com.ning.http.client.Response;
+import org.testng.annotations.Test;
+
+import java.util.concurrent.Executors;
+import java.util.concurrent.atomic.AtomicBoolean;
+
+import static org.testng.Assert.assertEquals;
+import static org.testng.Assert.assertNotNull;
+import static org.testng.Assert.assertTrue;
+
+/**
+ * Tests case where response doesn't have body.
+ *
+ * @author Hubert Iwaniuk
+ */
+public abstract class ListenableFutureTest extends AbstractBasicTest {
+
+    @Test(groups = {"standalone", "default_provider"})
+    public void testPutEmptyBody() throws Throwable {
+        final AtomicBoolean executed = new AtomicBoolean(false);
+        AsyncHttpClient ahc = getAsyncHttpClient(null);
+        Response response = ((ListenableFuture<Response>)ahc.prepareGet(getTargetUrl()).execute()).addListener(new Runnable(){
+
+
+            public void run() {
+                executed.set(true);
+            }
+        }, Executors.newFixedThreadPool(1)).get();
+
+        assertNotNull(response);
+        assertEquals(response.getStatusCode(), 200);
+        assertTrue(executed.get());
+        ahc.close();
+    }
+}
diff --git a/src/test/java/com/ning/http/client/async/netty/NettyListenableFutureTest.java b/src/test/java/com/ning/http/client/async/netty/NettyListenableFutureTest.java
@@ -0,0 +1,27 @@
+/*
+ * Copyright (c) 2010-2011 Sonatype, Inc. All rights reserved.
+ *
+ * This program is licensed to you under the Apache License Version 2.0,
+ * and you may not use this file except in compliance with the Apache License Version 2.0.
+ * You may obtain a copy of the Apache License Version 2.0 at http://www.apache.org/licenses/LICENSE-2.0.
+ *
+ * Unless required by applicable law or agreed to in writing,
+ * software distributed under the Apache License Version 2.0 is distributed on an
+ * "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the Apache License Version 2.0 for the specific language governing permissions and limitations there under.
+ */
+package com.ning.http.client.async.netty;
+
+import com.ning.http.client.AsyncHttpClient;
+import com.ning.http.client.AsyncHttpClientConfig;
+import com.ning.http.client.async.ListenableFutureTest;
+import com.ning.http.client.async.ProviderUtil;
+
+public class NettyListenableFutureTest extends ListenableFutureTest {
+
+    @Override
+    public AsyncHttpClient getAsyncHttpClient(AsyncHttpClientConfig config) {
+        return ProviderUtil.nettyProvider(config);
+    }
+
+}
