Adding extra support for rxjava

Author: drmaas

extras/pom.xml

@@ -45,6 +45,7 @@
         <module>guava</module>
         <module>jdeferred</module>
         <module>registry</module>
+        <module>rxjava</module>
     </modules>
 
     <dependencies>
@@ -61,4 +62,4 @@
             <classifier>tests</classifier>
         </dependency>
     </dependencies>
-</project>
+</project>

extras/rxjava/pom.xml

@@ -0,0 +1,31 @@
+<project xmlns="http://maven.apache.org/POM/4.0.0" xmlns:xsi="http://www.w3.org/2001/XMLSchema-instance"
+         xsi:schemaLocation="http://maven.apache.org/POM/4.0.0 http://maven.apache.org/xsd/maven-4.0.0.xsd">
+    <modelVersion>4.0.0</modelVersion>
+    <parent>
+        <artifactId>async-http-client-extras-parent</artifactId>
+        <groupId>org.asynchttpclient</groupId>
+        <version>2.0.0-SNAPSHOT</version>
+    </parent>
+    <artifactId>async-http-client-extras-rxjava</artifactId>
+    <name>Asynchronous Http Client RxJava Extras</name>
+    <description>The Async Http Client RxJava Extras.</description>
+    <dependencies>
+        <dependency>
+            <groupId>io.reactivex</groupId>
+            <artifactId>rxjava</artifactId>
+            <version>1.0.14</version>
+        </dependency>
+        <dependency>
+            <groupId>org.asynchttpclient</groupId>
+            <artifactId>async-http-client-netty3</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+        <dependency>
+            <groupId>org.asynchttpclient</groupId>
+            <artifactId>async-http-client-netty4</artifactId>
+            <version>${project.version}</version>
+            <scope>test</scope>
+        </dependency>
+    </dependencies>
+</project>

extras/rxjava/src/main/java/org/asynchttpclient/extras/rxjava/AsyncHttpClientErrorException.java

@@ -0,0 +1,15 @@
+package org.asynchttpclient.extras.rxjava;
+
+/**
+ *
+ */
+public class AsyncHttpClientErrorException extends RuntimeException {
+
+    public AsyncHttpClientErrorException(String message) {
+        super(message);
+    }
+
+    public AsyncHttpClientErrorException(String message, Throwable t) {
+        super(message, t);
+    }
+}

extras/rxjava/src/main/java/org/asynchttpclient/extras/rxjava/AsyncHttpObservable.java

@@ -0,0 +1,109 @@
+package org.asynchttpclient.extras.rxjava;
+
+import org.asynchttpclient.AsyncCompletionHandler;
+import org.asynchttpclient.BoundRequestBuilder;
+import org.asynchttpclient.HttpResponseStatus;
+import org.asynchttpclient.Response;
+import rx.Observable;
+import rx.Subscriber;
+import rx.functions.Func0;
+import rx.subjects.ReplaySubject;
+
+/**
+ * Provide RxJava support for executing requests. Request can be subscribed to and manipulated as needed.
+ * @See <a href="https://github.com/ReactiveX/RxJava" />
+ */
+public class AsyncHttpObservable {
+
+    /**
+     * Observe a request execution and emit the full response no matter what.
+     *
+     * @param supplier
+     * @return The cold observable (must be subscribed to in order to execute).
+     */
+    public static Observable<Response> toObservable(final Func0<BoundRequestBuilder> supplier) {
+        return toObservable(false, supplier);
+    }
+
+    /**
+     * Observe a request execution and emit an error for http status error codes >= 400.
+     *
+     * @param abortOnErrorStatus
+     * @param supplier
+     * @return The cold observable (must be subscribed to in order to execute).
+     */
+    public static Observable<Response> toObservable(final Boolean abortOnErrorStatus, final Func0<BoundRequestBuilder> supplier) {
+
+        //Get the builder from the function
+        final BoundRequestBuilder builder = supplier.call();
+
+        //create the observable from scratch
+        return Observable.create(new Observable.OnSubscribe<Response>() {
+
+            @Override
+            public void call(final Subscriber<? super Response> subscriber) {
+                try {
+                    AsyncCompletionHandler<Void> handler = new AsyncCompletionHandler<Void>() {
+                        @Override
+                        public State onStatusReceived(HttpResponseStatus status) throws Exception {
+                            State state = super.onStatusReceived(status);
+                            if (abortOnErrorStatus) {
+                                int code = status.getStatusCode();
+                                if (code >= 400) {
+                                    state = State.ABORT;
+                                    subscriber.onError(new AsyncHttpClientErrorException(String.format("Client error status code: %s", code)));
+                                }
+                            }
+                            return state;
+                        }
+
+                        @Override
+                        public Void onCompleted(Response response) throws Exception {
+                            subscriber.onNext(response);
+                            subscriber.onCompleted();
+                            return null;
+                        }
+
+                        @Override
+                        public void onThrowable(Throwable t) {
+                            subscriber.onError(t);
+                        }
+                    };
+                    //execute the request
+                    builder.execute(handler);
+                } catch (Throwable t) {
+                    subscriber.onError(t);
+                }
+            }
+
+        });
+
+    }
+
+    /**
+     * Observe a request execution and emit the full response no matter what.
+     *
+     * @param supplier
+     * @return The hot observable (eagerly executes).
+     */
+    public static Observable<Response> observe(final Func0<BoundRequestBuilder> supplier) {
+        return observe(false, supplier);
+    }
+
+    /**
+     * Observe a request execution and emit an error for http status error codes >= 400.
+     *
+     * @param abortOnErrorStatus
+     * @param supplier
+     * @return The hot observable (eagerly executes).
+     */
+    public static Observable<Response> observe(final Boolean abortOnErrorStatus, final Func0<BoundRequestBuilder> supplier) {
+        //use a ReplaySubject to buffer the eagerly subscribed-to Observable
+        ReplaySubject<Response> subject = ReplaySubject.create();
+        //eagerly kick off subscription
+        toObservable(abortOnErrorStatus, supplier).subscribe(subject);
+        //return the subject that can be subscribed to later while the execution has already started
+        return subject;
+    }
+
+}