From 3f279ef47848c3670ae68cd4501b94c273d51008 Mon Sep 17 00:00:00 2001
From: Vishesh Vadhera <vishesh.2710@gmail.com>
Date: Fri, 14 Apr 2017 04:45:54 +0530
Subject: [PATCH] Rx Completable for ApolloPrefetch (#408)

---
 .../apollographql/apollo/RxApolloTest.java    | 36 +++++++++++++++-
 .../RxApollo.java                             | 42 +++++++++++++++++++
 2 files changed, 77 insertions(+), 1 deletion(-)

diff --git a/apollo-integration/src/test/java/com/apollographql/apollo/RxApolloTest.java b/apollo-integration/src/test/java/com/apollographql/apollo/RxApolloTest.java
index 8d699a96e04..07e9e05d06f 100644
--- a/apollo-integration/src/test/java/com/apollographql/apollo/RxApolloTest.java
+++ b/apollo-integration/src/test/java/com/apollographql/apollo/RxApolloTest.java
@@ -29,13 +29,17 @@
 import okhttp3.mockwebserver.MockWebServer;
 import rx.Observer;
 import rx.Subscription;
+import rx.observers.AssertableSubscriber;
 import rx.observers.TestSubscriber;
 
 import static com.google.common.truth.Truth.assertThat;
 
 public class RxApolloTest {
+
   private ApolloClient apolloClient;
   private MockWebServer server;
+
+  private static final int RX_DELAY_SECONDS = 2;
   private static final long TIME_OUT_SECONDS = 3;
 
   @Before public void setUp() {
@@ -88,7 +92,7 @@ public class RxApolloTest {
     TestSubscriber<EpisodeHeroName.Data> subscriber = new TestSubscriber<>();
     Subscription subscription = RxApollo
         .from(call)
-        .delay(5, TimeUnit.SECONDS)
+        .delay(RX_DELAY_SECONDS, TimeUnit.SECONDS)
         .subscribe(subscriber);
     subscription.unsubscribe();
     subscriber
@@ -97,6 +101,36 @@ public class RxApolloTest {
 
   }
 
+  @Test public void testRxPrefetchCompletes() throws IOException {
+    EpisodeHeroName query = EpisodeHeroName.builder().episode(Episode.EMPIRE).build();
+    server.enqueue(mockResponse("EpisodeHeroNameResponseWithId.json"));
+
+    RxApollo
+        .from(apolloClient.prefetch(query))
+        .test()
+        .awaitTerminalEvent()
+        .assertNoErrors()
+        .assertCompleted();
+  }
+
+  @Test public void testRxPrefetchIsCanceledWhenUnsubscribed() throws IOException {
+
+    EpisodeHeroName query = EpisodeHeroName.builder().episode(Episode.EMPIRE).build();
+    server.enqueue(mockResponse("EpisodeHeroNameResponseWithId.json"));
+
+    ApolloPrefetch prefetch = apolloClient.prefetch(query);
+
+    AssertableSubscriber<Void> subscriber = RxApollo
+        .from(prefetch)
+        .delay(RX_DELAY_SECONDS, TimeUnit.SECONDS)
+        .test();
+
+    subscriber.unsubscribe();
+
+    subscriber.assertUnsubscribed();
+    subscriber.assertNotCompleted();
+  }
+
   @Test
   public void testRxQueryWatcherUpdated_SameQuery_DifferentResults() throws IOException, InterruptedException,
       TimeoutException, ApolloException {
diff --git a/apollo-rxsupport/src/main/java/com.apollographql.android.rx/RxApollo.java b/apollo-rxsupport/src/main/java/com.apollographql.android.rx/RxApollo.java
index 45bbdb29380..987a21dd627 100644
--- a/apollo-rxsupport/src/main/java/com.apollographql.android.rx/RxApollo.java
+++ b/apollo-rxsupport/src/main/java/com.apollographql.android.rx/RxApollo.java
@@ -2,6 +2,7 @@
 
 
 import com.apollographql.apollo.ApolloCall;
+import com.apollographql.apollo.ApolloPrefetch;
 import com.apollographql.apollo.ApolloWatcher;
 import com.apollographql.apollo.api.Response;
 import com.apollographql.apollo.exception.ApolloException;
@@ -9,10 +10,13 @@
 
 import javax.annotation.Nonnull;
 
+import rx.Completable;
+import rx.CompletableSubscriber;
 import rx.Emitter;
 import rx.Observable;
 import rx.Single;
 import rx.SingleSubscriber;
+import rx.Subscription;
 import rx.exceptions.Exceptions;
 import rx.functions.Action0;
 import rx.functions.Action1;
@@ -103,6 +107,44 @@ public static <T> Observable<T> from(@Nonnull final ApolloWatcher<T> watcher) {
     });
   }
 
+  /**
+   * Converts an {@link ApolloPrefetch} to a Completable.
+   *
+   * @param prefetch the ApolloPrefetch to convert
+   * @return the converted Completable
+   */
+  @Nonnull public static Completable from(@Nonnull final ApolloPrefetch prefetch) {
+    checkNotNull(prefetch, "prefetch == null");
+
+    return Completable.create(new Completable.OnSubscribe() {
+      @Override public void call(final CompletableSubscriber subscriber) {
+        Subscription subscription = getSubscription(subscriber, prefetch);
+
+        try {
+          prefetch.execute();
+          if (!subscription.isUnsubscribed()) {
+            subscriber.onCompleted();
+          }
+        } catch (ApolloException e) {
+          Exceptions.throwIfFatal(e);
+          if (!subscription.isUnsubscribed()) {
+            subscriber.onError(e);
+          }
+        }
+      }
+    });
+  }
+
+  private static Subscription getSubscription(CompletableSubscriber subscriber, final Cancelable cancelable) {
+    Subscription subscription = Subscriptions.create(new Action0() {
+      @Override public void call() {
+        cancelable.cancel();
+      }
+    });
+    subscriber.onSubscribe(subscription);
+    return subscription;
+  }
+
   private static <T> void cancelOnSingleUnsubscribe(SingleSubscriber<? super T> subscriber, final Cancelable toCancel) {
     subscriber.add(Subscriptions.create(new Action0() {
       @Override public void call() {