### Question:
AsyncHttpObservable { public static Observable<Response> observe(final Func0<BoundRequestBuilder> supplier) { ReplaySubject<Response> subject = ReplaySubject.create(); toObservable(supplier).subscribe(subject); return subject; }  static Observable<Response> toObservable(final Func0<BoundRequestBuilder> supplier); static Observable<Response> observe(final Func0<BoundRequestBuilder> supplier);  }### Answer:
@Test public void testObserveNoError() { final TestSubscriber<Response> tester = new TestSubscriber<>(); try (AsyncHttpClient client = asyncHttpClient()) { Observable<Response> o1 = AsyncHttpObservable.observe(() -> client.prepareGet("https: o1.subscribe(tester); tester.awaitTerminalEvent(); tester.assertTerminalEvent(); tester.assertNoErrors(); tester.assertCompleted(); List<Response> responses = tester.getOnNextEvents(); assertNotNull(responses); assertEquals(responses.size(), 1); assertEquals(responses.get(0).getStatusCode(), 200); } catch (Exception e) { Thread.currentThread().interrupt(); } }
@Test public void testObserveError() { final TestSubscriber<Response> tester = new TestSubscriber<>(); try (AsyncHttpClient client = asyncHttpClient()) { Observable<Response> o1 = AsyncHttpObservable.observe(() -> client.prepareGet("https: o1.subscribe(tester); tester.awaitTerminalEvent(); tester.assertTerminalEvent(); tester.assertNoErrors(); tester.assertCompleted(); List<Response> responses = tester.getOnNextEvents(); assertNotNull(responses); assertEquals(responses.size(), 1); assertEquals(responses.get(0).getStatusCode(), 404); } catch (Exception e) { Thread.currentThread().interrupt(); } }
@Test public void testObserveMultiple() { final TestSubscriber<Response> tester = new TestSubscriber<>(); try (AsyncHttpClient client = asyncHttpClient()) { Observable<Response> o1 = AsyncHttpObservable.observe(() -> client.prepareGet("https: Observable<Response> o2 = AsyncHttpObservable.observe(() -> client.prepareGet("http: Observable<Response> o3 = AsyncHttpObservable.observe(() -> client.prepareGet("http: Observable<Response> all = Observable.merge(o1, o2, o3); all.subscribe(tester); tester.awaitTerminalEvent(); tester.assertTerminalEvent(); tester.assertNoErrors(); tester.assertCompleted(); List<Response> responses = tester.getOnNextEvents(); assertNotNull(responses); assertEquals(responses.size(), 3); for (Response response : responses) { assertEquals(response.getStatusCode(), 200); } } catch (Exception e) { Thread.currentThread().interrupt(); } }