Removing custom abort logic and custom exception

Author: drmaas

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

@@ -28,23 +28,12 @@
 public class AsyncHttpObservable {
 
     /**
-     * Observe a request execution and emit the full response no matter what.
+     * Observe a request execution and emit the response to the observer.
      *
      * @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();
@@ -58,23 +47,13 @@ public void call(final Subscriber<? super Response> subscriber) {
                     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;
+                            return super.onStatusReceived(status);
                         }
 
                         @Override
                         public Void onCompleted(Response response) throws Exception {
-                            if (!(abortOnErrorStatus && response.getStatusCode() >= 400)) {
-                                subscriber.onNext(response);
-                                subscriber.onCompleted();
-                            }
+                            subscriber.onNext(response);
+                            subscriber.onCompleted();
                             return null;
                         }
 
@@ -95,27 +74,16 @@ public void onThrowable(Throwable t) {
     }
 
     /**
-     * Observe a request execution and emit the full response no matter what.
+     * Observe a request execution and emit the response to the observer.
      *
      * @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);
+        toObservable(supplier).subscribe(subject);
         //return the subject that can be subscribed to later while the execution has already started
         return subject;
     }

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

@@ -32,7 +32,7 @@
 public class AsyncHttpObservableTest {
 
     @Test(groups = "fast")
-    public void testToObservableNoAbortNoError() {
+    public void testToObservableNoError() {
         final TestSubscriber<Response> tester = new TestSubscriber<>();
 
         try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
@@ -57,7 +57,7 @@ public BoundRequestBuilder call() {
     }
 
     @Test(groups = "fast")
-    public void testToObservableNoAbortError() {
+    public void testToObservableError() {
         final TestSubscriber<Response> tester = new TestSubscriber<>();
 
         try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
@@ -82,55 +82,7 @@ public BoundRequestBuilder call() {
     }
 
     @Test(groups = "fast")
-    public void testToObservableAbortNoError() {
-        final TestSubscriber<Response> tester = new TestSubscriber<>();
-
-        try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
-            Observable<Response> o1 = AsyncHttpObservable.toObservable(true, new Func0<BoundRequestBuilder>() {
-                @Override
-                public BoundRequestBuilder call() {
-                    return client.prepareGet("http://www.ning.com");
-                }
-            });
-            o1.subscribe(tester);
-            tester.awaitTerminalEvent();
-            tester.assertTerminalEvent();
-            tester.assertCompleted();
-            List<Response> responses = tester.getOnNextEvents();
-            assertNotNull(responses);
-            assertEquals(responses.size(), 1);
-            assertEquals(responses.get(1).getStatusCode(), 200);
-        } catch (Exception e) {
-            Thread.currentThread().interrupt();
-        }
-    }
-
-    @Test(groups = "fast")
-    public void testToObservableAbortError() {
-        final TestSubscriber<Response> tester = new TestSubscriber<>();
-
-        try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
-            Observable<Response> o1 = AsyncHttpObservable.toObservable(true, new Func0<BoundRequestBuilder>() {
-                @Override
-                public BoundRequestBuilder call() {
-                    return client.prepareGet("http://www.ning.com/ttfn");
-                }
-            });
-            o1.subscribe(tester);
-            tester.awaitTerminalEvent();
-            tester.assertTerminalEvent();
-            tester.assertNotCompleted();
-            tester.assertError(AsyncHttpClientErrorException.class);
-            List<Response> responses = tester.getOnNextEvents();
-            assertNotNull(responses);
-            assertEquals(responses.size(), 0);
-        } catch (Exception e) {
-            Thread.currentThread().interrupt();
-        }
-    }
-
-    @Test(groups = "fast")
-    public void testObserveNoAbortNoError() {
+    public void testObserveNoError() {
         final TestSubscriber<Response> tester = new TestSubscriber<>();
 
         try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
@@ -155,7 +107,7 @@ public BoundRequestBuilder call() {
     }
 
     @Test(groups = "fast")
-    public void testObserveNoAbortError() {
+    public void testObserveError() {
         final TestSubscriber<Response> tester = new TestSubscriber<>();
 
         try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
@@ -179,54 +131,6 @@ public BoundRequestBuilder call() {
         }
     }
 
-    @Test(groups = "fast")
-    public void testObserveAbortNoError() {
-        final TestSubscriber<Response> tester = new TestSubscriber<>();
-
-        try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
-            Observable<Response> o1 = AsyncHttpObservable.observe(true, new Func0<BoundRequestBuilder>() {
-                @Override
-                public BoundRequestBuilder call() {
-                    return client.prepareGet("http://www.ning.com");
-                }
-            });
-            o1.subscribe(tester);
-            tester.awaitTerminalEvent();
-            tester.assertTerminalEvent();
-            tester.assertCompleted();
-            List<Response> responses = tester.getOnNextEvents();
-            assertNotNull(responses);
-            assertEquals(responses.size(), 1);
-            assertEquals(responses.get(1).getStatusCode(), 200);
-        } catch (Exception e) {
-            Thread.currentThread().interrupt();
-        }
-    }
-
-    @Test(groups = "fast")
-    public void testObserveAbortError() {
-        final TestSubscriber<Response> tester = new TestSubscriber<>();
-
-        try (AsyncHttpClient client = new DefaultAsyncHttpClient()) {
-            Observable<Response> o1 = AsyncHttpObservable.observe(true, new Func0<BoundRequestBuilder>() {
-                @Override
-                public BoundRequestBuilder call() {
-                    return client.prepareGet("http://www.ning.com/ttfn");
-                }
-            });
-            o1.subscribe(tester);
-            tester.awaitTerminalEvent();
-            tester.assertTerminalEvent();
-            tester.assertNotCompleted();
-            tester.assertError(AsyncHttpClientErrorException.class);
-            List<Response> responses = tester.getOnNextEvents();
-            assertNotNull(responses);
-            assertEquals(responses.size(), 0);
-        } catch (Exception e) {
-            Thread.currentThread().interrupt();
-        }
-    }
-
     @Test(groups = "fast")
     public void testObserveMultiple() {
         final TestSubscriber<Response> tester = new TestSubscriber<>();