Fix asFlow() leak (#2164) (#2175)

apollographql · Apr 16, 2020 · 744365e · 744365e
1 parent b15f11b
commit 744365e
Show file tree

Hide file tree

Showing 2 changed files with 34 additions and 6 deletions.
diff --git a/...tines-support/src/main/kotlin/com/apollographql/apollo/coroutines/CoroutinesExtensions.kt b/...tines-support/src/main/kotlin/com/apollographql/apollo/coroutines/CoroutinesExtensions.kt
@@ -23,7 +23,8 @@ import kotlinx.coroutines.flow.callbackFlow
  */
 @ExperimentalCoroutinesApi
 fun <T> ApolloCall<T>.toFlow(): Flow<Response<T>> = callbackFlow {
-  clone().enqueue(
+  val clone = clone()
+  clone.enqueue(
       object : ApolloCall.Callback<T>() {
         override fun onResponse(response: Response<T>) {
           runCatching {
@@ -42,7 +43,7 @@ fun <T> ApolloCall<T>.toFlow(): Flow<Response<T>> = callbackFlow {
         }
       }
   )
-  awaitClose { this@toFlow.cancel() }
+  awaitClose { clone.cancel() }
 }
 
 /**
@@ -53,7 +54,8 @@ fun <T> ApolloCall<T>.toFlow(): Flow<Response<T>> = callbackFlow {
  */
 @ExperimentalCoroutinesApi
 fun <T> ApolloQueryWatcher<T>.toFlow(): Flow<Response<T>> = callbackFlow {
-  clone().enqueueAndWatch(
+  val clone = clone()
+  clone.enqueueAndWatch(
       object : ApolloCall.Callback<T>() {
         override fun onResponse(response: Response<T>) {
           runCatching {
@@ -72,7 +74,7 @@ fun <T> ApolloQueryWatcher<T>.toFlow(): Flow<Response<T>> = callbackFlow {
         }
       }
   )
-  awaitClose { this@toFlow.cancel() }
+  awaitClose { clone.cancel() }
 }
 
 
@@ -116,7 +118,8 @@ fun <T> ApolloCall<T>.toDeferred(): Deferred<Response<T>> {
  */
 @ExperimentalCoroutinesApi
 fun <T> ApolloSubscriptionCall<T>.toFlow(): Flow<Response<T>> = callbackFlow {
-  clone().execute(
+  val clone = clone()
+  clone.execute(
       object : ApolloSubscriptionCall.Callback<T> {
         override fun onConnected() {
         }
@@ -140,7 +143,7 @@ fun <T> ApolloSubscriptionCall<T>.toFlow(): Flow<Response<T>> = callbackFlow {
         }
       }
   )
-  awaitClose { this@toFlow.cancel() }
+  awaitClose { clone.cancel() }
 }
 
 /**

diff --git a/apollo-integration/src/test/java/com/apollographql/apollo/CoroutinesApolloTest.kt b/apollo-integration/src/test/java/com/apollographql/apollo/CoroutinesApolloTest.kt
@@ -12,6 +12,8 @@ import com.apollographql.apollo.coroutines.toFlow
 import com.apollographql.apollo.coroutines.toJob
 import com.apollographql.apollo.exception.ApolloException
 import com.apollographql.apollo.integration.normalizer.EpisodeHeroNameQuery
+import com.apollographql.apollo.integration.normalizer.HeroNameQuery
+import com.apollographql.apollo.integration.normalizer.HeroNameWithIdQuery
 import com.apollographql.apollo.integration.normalizer.type.Episode
 import com.google.common.truth.Truth.assertThat
 import kotlinx.coroutines.ExperimentalCoroutinesApi
@@ -149,6 +151,29 @@ class CoroutinesApolloTest {
     assertThat(response.data!!.hero()!!.name()).isEqualTo("R2-D2")
   }
 
+  @Test
+  @ExperimentalCoroutinesApi
+  fun watcherFlowCancellationCancelsWatcher(): Unit = runBlocking {
+    server.enqueue(mockResponse("HeroNameWithIdResponse.json"))
+    apolloClient
+        .query(HeroNameWithIdQuery())
+        .watcher()
+        .toFlow()
+        .first() // Cancels the flow after first response
+
+    apolloClient.clearNormalizedCache()
+    apolloClient.clearHttpCache()
+
+    server.enqueue(mockResponse("HeroNameResponse.json"))
+    apolloClient
+        .query(HeroNameQuery())
+        .watcher()
+        .toFlow()
+        .first()
+
+    assertThat(server.requestCount).isEqualTo(2)
+  }.let { }
+
   companion object {
 
     private const val FILE_EPISODE_HERO_NAME_WITH_ID = "EpisodeHeroNameResponseWithId.json"