Merge pull request #1980 from square/jw/rex

Switch to a better, custom validating subscriber.

pom.xml

@@ -53,7 +53,7 @@
     <animal.sniffer.version>1.14</animal.sniffer.version>
 
     <!-- Adapter Dependencies -->
-    <rxjava.version>1.1.5</rxjava.version>
+    <rxjava.version>1.1.9</rxjava.version>
 
     <!-- Converter Dependencies -->
     <gson.version>2.7</gson.version>

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/CompletableTest.java

@@ -24,13 +24,13 @@ import org.junit.Test;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Completable;
-import rx.observers.TestSubscriber;
 
 import static okhttp3.mockwebserver.SocketPolicy.DISCONNECT_AFTER_REQUEST;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public final class CompletableTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Completable completable();
@@ -49,25 +49,24 @@ public final class CompletableTest {
   @Test public void completableSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Void> subscriber = new TestSubscriber<>();
-    service.completable().subscribe(subscriber);
+    RecordingSubscriber<Void> subscriber = subscriberRule.create();
+    service.completable().unsafeSubscribe(subscriber);
     subscriber.assertCompleted();
   }
 
   @Test public void completableSuccess404() {
     server.enqueue(new MockResponse().setResponseCode(404));
 
-    TestSubscriber<Void> subscriber = new TestSubscriber<>();
-    service.completable().subscribe(subscriber);
-    Throwable cause = subscriber.getOnErrorEvents().get(0);
-    assertThat(cause).isInstanceOf(HttpException.class).hasMessage("HTTP 404 Client Error");
+    RecordingSubscriber<Void> subscriber = subscriberRule.create();
+    service.completable().unsafeSubscribe(subscriber);
+    subscriber.assertError(HttpException.class, "HTTP 404 Client Error");
   }
 
   @Test public void completableFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<Void> subscriber = new TestSubscriber<>();
-    service.completable().subscribe(subscriber);
+    RecordingSubscriber<Void> subscriber = subscriberRule.create();
+    service.completable().unsafeSubscribe(subscriber);
     subscriber.assertError(IOException.class);
   }
 }

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/CompletableWithSchedulerTest.java

@@ -23,11 +23,11 @@ import org.junit.Test;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Completable;
-import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
 
 public final class CompletableWithSchedulerTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Completable completable();
@@ -47,9 +47,9 @@ public final class CompletableWithSchedulerTest {
   @Test public void completableUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Void> subscriber = new TestSubscriber<>();
-    service.completable().subscribe(subscriber);
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.completable().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
     subscriber.assertCompleted();

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/ObservableTest.java

@@ -25,13 +25,13 @@ import retrofit2.Response;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Observable;
-import rx.observers.TestSubscriber;
 
 import static okhttp3.mockwebserver.SocketPolicy.DISCONNECT_AFTER_REQUEST;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public final class ObservableTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Observable<String> body();
@@ -53,42 +53,37 @@ public final class ObservableTest {
   @Test public void bodySuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertValues("Hi");
-    subscriber.assertCompleted();
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertValue("Hi").assertCompleted();
   }
 
   @Test public void bodySuccess404() {
     server.enqueue(new MockResponse().setResponseCode(404));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertNoValues();
-    Throwable cause = subscriber.getOnErrorEvents().get(0);
-    assertThat(cause).isInstanceOf(HttpException.class).hasMessage("HTTP 404 Client Error");
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertError(HttpException.class, "HTTP 404 Client Error");
   }
 
   @Test public void bodyFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertNoValues();
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
     subscriber.assertError(IOException.class);
   }
 
   @Test public void bodyRespectsBackpressure() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>(0);
-    Observable<String> o = service.body();
-
-    o.subscribe(subscriber);
+    RecordingSubscriber<String> subscriber = subscriberRule.createWithInitialRequest(0);
+    service.body().unsafeSubscribe(subscriber);
     assertThat(server.getRequestCount()).isEqualTo(0);
 
     subscriber.requestMore(1);
     assertThat(server.getRequestCount()).isEqualTo(1);
+    subscriber.assertAnyValue().assertCompleted();
 
     subscriber.requestMore(Long.MAX_VALUE); // Subsequent requests do not trigger HTTP requests.
     assertThat(server.getRequestCount()).isEqualTo(1);
@@ -97,45 +92,39 @@ public final class ObservableTest {
   @Test public void responseSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    Response<String> response = subscriber.getOnNextEvents().get(0);
-    assertThat(response.isSuccessful()).isTrue();
-    assertThat(response.body()).isEqualTo("Hi");
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().body()).isEqualTo("Hi");
     subscriber.assertCompleted();
   }
 
   @Test public void responseSuccess404() throws IOException {
-    server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
+    server.enqueue(new MockResponse().setResponseCode(404));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    Response<String> response = subscriber.getOnNextEvents().get(0);
-    assertThat(response.isSuccessful()).isFalse();
-    assertThat(response.errorBody().string()).isEqualTo("Hi");
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().code()).isEqualTo(404);
     subscriber.assertCompleted();
   }
 
   @Test public void responseFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    subscriber.assertNoValues();
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
     subscriber.assertError(IOException.class);
   }
 
   @Test public void responseRespectsBackpressure() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>(0);
-    Observable<Response<String>> o = service.response();
-
-    o.subscribe(subscriber);
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.createWithInitialRequest(0);
+    service.response().unsafeSubscribe(subscriber);
     assertThat(server.getRequestCount()).isEqualTo(0);
 
     subscriber.requestMore(1);
     assertThat(server.getRequestCount()).isEqualTo(1);
+    subscriber.assertAnyValue().assertCompleted();
 
     subscriber.requestMore(Long.MAX_VALUE); // Subsequent requests do not trigger HTTP requests.
     assertThat(server.getRequestCount()).isEqualTo(1);
@@ -144,51 +133,40 @@ public final class ObservableTest {
   @Test public void resultSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isFalse();
-    Response<String> response = result.response();
-    assertThat(response.isSuccessful()).isTrue();
-    assertThat(response.body()).isEqualTo("Hi");
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().response().body()).isEqualTo("Hi");
     subscriber.assertCompleted();
   }
 
   @Test public void resultSuccess404() throws IOException {
-    server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
-
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isFalse();
-    Response<String> response = result.response();
-    assertThat(response.isSuccessful()).isFalse();
-    assertThat(response.errorBody().string()).isEqualTo("Hi");
+    server.enqueue(new MockResponse().setResponseCode(404));
+
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().response().code()).isEqualTo(404);
     subscriber.assertCompleted();
   }
 
   @Test public void resultFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isTrue();
-    assertThat(result.error()).isInstanceOf(IOException.class);
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().error()).isInstanceOf(IOException.class);
     subscriber.assertCompleted();
   }
 
   @Test public void resultRespectsBackpressure() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>(0);
-    Observable<Result<String>> o = service.result();
-
-    o.subscribe(subscriber);
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.createWithInitialRequest(0);
+    service.result().unsafeSubscribe(subscriber);
     assertThat(server.getRequestCount()).isEqualTo(0);
 
     subscriber.requestMore(1);
     assertThat(server.getRequestCount()).isEqualTo(1);
+    subscriber.assertAnyValue().assertCompleted();
 
     subscriber.requestMore(Long.MAX_VALUE); // Subsequent requests do not trigger HTTP requests.
     assertThat(server.getRequestCount()).isEqualTo(1);

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/ObservableWithSchedulerTest.java

@@ -24,11 +24,11 @@ import retrofit2.Response;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Observable;
-import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
 
 public final class ObservableWithSchedulerTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Observable<String> body();
@@ -51,39 +51,33 @@ public final class ObservableWithSchedulerTest {
   @Test public void bodyUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
-    subscriber.assertCompleted();
+    subscriber.assertAnyValue().assertCompleted();
   }
 
   @Test public void responseUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
-    subscriber.assertCompleted();
+    subscriber.assertAnyValue().assertCompleted();
   }
 
   @Test public void resultUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
-    subscriber.assertCompleted();
+    subscriber.assertAnyValue().assertCompleted();
   }
 }

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/RecordingSubscriber.java

+/*
+ * Copyright (C) 2016 Square, 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 retrofit2.adapter.rxjava;
+
+import java.util.ArrayDeque;
+import java.util.ArrayList;
+import java.util.Deque;
+import java.util.List;
+import org.junit.rules.TestRule;
+import org.junit.runner.Description;
+import org.junit.runners.model.Statement;
+import rx.Notification;
+import rx.Subscriber;
+
+import static org.assertj.core.api.Assertions.assertThat;
+
+/** A test {@link Subscriber} and JUnit rule which guarantees all events are asserted. */
+final class RecordingSubscriber<T> extends Subscriber<T> {
+  private final long initialRequest;
+  private final Deque<Notification<T>> events = new ArrayDeque<>();
+
+  private RecordingSubscriber(long initialRequest) {
+    this.initialRequest = initialRequest;
+  }
+
+  @Override public void onStart() {
+    request(initialRequest);
+  }
+
+  @Override public void onNext(T value) {
+    events.add(Notification.createOnNext(value));
+  }
+
+  @Override public void onCompleted() {
+    events.add(Notification.<T>createOnCompleted());
+  }
+
+  @Override public void onError(Throwable e) {
+    events.add(Notification.<T>createOnError(e));
+  }
+
+  private Notification<T> takeNotification() {
+    Notification<T> notification = events.pollFirst();
+    if (notification == null) {
+      throw new AssertionError("No event found!");
+    }
+    return notification;
+  }
+
+  public T takeValue() {
+    Notification<T> notification = takeNotification();
+    assertThat(notification.isOnNext())
+        .overridingErrorMessage("Expected onNext event but was %s", notification)
+        .isTrue();
+    return notification.getValue();
+  }
+
+  public Throwable takeError() {
+    Notification<T> notification = takeNotification();
+    assertThat(notification.isOnError())
+        .overridingErrorMessage("Expected onError event but was %s", notification)
+        .isTrue();
+    return notification.getThrowable();
+  }
+
+  public RecordingSubscriber<T> assertAnyValue() {
+    takeValue();
+    return this;
+  }
+
+  public RecordingSubscriber<T> assertValue(T value) {
+    assertThat(takeValue()).isEqualTo(value);
+    return this;
+  }
+
+  public void assertCompleted() {
+    Notification<T> notification = takeNotification();
+    assertThat(notification.isOnCompleted())
+        .overridingErrorMessage("Expected onCompleted event but was %s", notification)
+        .isTrue();
+    assertNoEvents();
+  }
+
+  public void assertError(Throwable throwable) {
+    assertThat(takeError()).isEqualTo(throwable);
+  }
+
+  public void assertError(Class<? extends Throwable> errorClass) {
+    assertError(errorClass, null);
+  }
+
+  public void assertError(Class<? extends Throwable> errorClass, String message) {
+    Throwable throwable = takeError();
+    assertThat(throwable).isInstanceOf(errorClass);
+    if (message != null) {
+      assertThat(throwable).hasMessage(message);
+    }
+    assertNoEvents();
+  }
+
+  public void assertNoEvents() {
+    assertThat(events).as("Unconsumed events found!").isEmpty();
+  }
+
+  public void requestMore(long amount) {
+    request(amount);
+  }
+
+  public static final class Rule implements TestRule {
+    final List<RecordingSubscriber<?>> subscribers = new ArrayList<>();
+
+    public <T> RecordingSubscriber<T> create() {
+      return createWithInitialRequest(Long.MAX_VALUE);
+    }
+
+    public <T> RecordingSubscriber<T> createWithInitialRequest(long initialRequest) {
+      RecordingSubscriber<T> subscriber = new RecordingSubscriber<>(initialRequest);
+      subscribers.add(subscriber);
+      return subscriber;
+    }
+
+    @Override public Statement apply(final Statement base, Description description) {
+      return new Statement() {
+        @Override public void evaluate() throws Throwable {
+          base.evaluate();
+          for (RecordingSubscriber<?> subscriber : subscribers) {
+            subscriber.assertNoEvents();
+          }
+        }
+      };
+    }
+  }
+}

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/SingleTest.java

@@ -25,13 +25,13 @@ import retrofit2.Response;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Single;
-import rx.observers.TestSubscriber;
 
 import static okhttp3.mockwebserver.SocketPolicy.DISCONNECT_AFTER_REQUEST;
 import static org.assertj.core.api.Assertions.assertThat;
 
 public final class SingleTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Single<String> body();
@@ -53,87 +53,77 @@ public final class SingleTest {
   @Test public void bodySuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertValue("Hi");
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertValue("Hi").assertCompleted();
   }
 
   @Test public void bodySuccess404() {
     server.enqueue(new MockResponse().setResponseCode(404));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    Throwable cause = subscriber.getOnErrorEvents().get(0);
-    assertThat(cause).isInstanceOf(HttpException.class).hasMessage("HTTP 404 Client Error");
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertError(HttpException.class, "HTTP 404 Client Error");
   }
 
   @Test public void bodyFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
     subscriber.assertError(IOException.class);
   }
 
   @Test public void responseSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    Response<String> response = subscriber.getOnNextEvents().get(0);
-    assertThat(response.isSuccessful()).isTrue();
-    assertThat(response.body()).isEqualTo("Hi");
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().body()).isEqualTo("Hi");
+    subscriber.assertCompleted();
   }
 
   @Test public void responseSuccess404() throws IOException {
-    server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
+    server.enqueue(new MockResponse().setResponseCode(404));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    Response<String> response = subscriber.getOnNextEvents().get(0);
-    assertThat(response.isSuccessful()).isFalse();
-    assertThat(response.errorBody().string()).isEqualTo("Hi");
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().code()).isEqualTo(404);
+    subscriber.assertCompleted();
   }
 
   @Test public void responseFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
     subscriber.assertError(IOException.class);
   }
 
   @Test public void resultSuccess200() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isFalse();
-    Response<String> response = result.response();
-    assertThat(response.isSuccessful()).isTrue();
-    assertThat(response.body()).isEqualTo("Hi");
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().response().body()).isEqualTo("Hi");
+    subscriber.assertCompleted();
   }
 
   @Test public void resultSuccess404() throws IOException {
-    server.enqueue(new MockResponse().setResponseCode(404).setBody("Hi"));
-
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isFalse();
-    Response<String> response = result.response();
-    assertThat(response.isSuccessful()).isFalse();
-    assertThat(response.errorBody().string()).isEqualTo("Hi");
+    server.enqueue(new MockResponse().setResponseCode(404));
+
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().response().code()).isEqualTo(404);
+    subscriber.assertCompleted();
   }
 
   @Test public void resultFailure() {
     server.enqueue(new MockResponse().setSocketPolicy(DISCONNECT_AFTER_REQUEST));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    Result<String> result = subscriber.getOnNextEvents().get(0);
-    assertThat(result.isError()).isTrue();
-    assertThat(result.error()).isInstanceOf(IOException.class);
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    assertThat(subscriber.takeValue().error()).isInstanceOf(IOException.class);
+    subscriber.assertCompleted();
   }
 }

retrofit-adapters/rxjava/src/test/java/retrofit2/adapter/rxjava/SingleWithSchedulerTest.java

@@ -24,11 +24,11 @@ import retrofit2.Response;
 import retrofit2.Retrofit;
 import retrofit2.http.GET;
 import rx.Single;
-import rx.observers.TestSubscriber;
 import rx.schedulers.TestScheduler;
 
 public final class SingleWithSchedulerTest {
   @Rule public final MockWebServer server = new MockWebServer();
+  @Rule public final RecordingSubscriber.Rule subscriberRule = new RecordingSubscriber.Rule();
 
   interface Service {
     @GET("/") Single<String> body();
@@ -51,36 +51,33 @@ public final class SingleWithSchedulerTest {
   @Test public void bodyUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<String> subscriber = new TestSubscriber<>();
-    service.body().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<String> subscriber = subscriberRule.create();
+    service.body().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
+    subscriber.assertAnyValue().assertCompleted();
   }
 
   @Test public void responseUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Response<String>> subscriber = new TestSubscriber<>();
-    service.response().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<Response<String>> subscriber = subscriberRule.create();
+    service.response().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
+    subscriber.assertAnyValue().assertCompleted();
   }
 
   @Test public void resultUsesScheduler() {
     server.enqueue(new MockResponse().setBody("Hi"));
 
-    TestSubscriber<Result<String>> subscriber = new TestSubscriber<>();
-    service.result().subscribe(subscriber);
-    subscriber.assertNoValues();
-    subscriber.assertNoTerminalEvent();
+    RecordingSubscriber<Result<String>> subscriber = subscriberRule.create();
+    service.result().unsafeSubscribe(subscriber);
+    subscriber.assertNoEvents();
 
     scheduler.triggerActions();
-    subscriber.assertValueCount(1);
+    subscriber.assertAnyValue().assertCompleted();
   }
 }