asFlow() leaks cloned watcher

[lwasyl]

Current implementation of asFlow is flawed in that it will leave the call active even when the flow's coroutine has been cancelled.

You can observe this behavior on my commit https://github.com/lwasyl/apollo-android/commit/9c6fde79844cf8c31e2f5c2b4e042c7b59ad6d30 in CoroutinesApolloTest#watcherFlowCancellationCancelsWatcher test. The test is failing but I left the logs to make it clear what's happening:

Awaiting first flow
[FoosCall] Got event SCHEDULED
[FoosCall] Got event FETCH_CACHE
[FoosCall] Got event FETCH_NETWORK
[FoosCall] Got response com.apollographql.apollo.api.Response@7e1bcf9c
[FoosCall] Got event COMPLETED
Cancelling first job
Clearing caches
Enqueueing second response
Calling second query
[BarsCall] Got event SCHEDULED
[BarsCall] Got event FETCH_NETWORK
[FoosCall] Got event SCHEDULED
[FoosCall] Got event FETCH_CACHE
[FoosCall] Got event FETCH_NETWORK 
[FoosCall] Got failure com.apollographql.apollo.exception.ApolloNetworkException: Failed to execute http call
[BarsCall] Got response com.apollographql.apollo.api.Response@cab19945
[BarsCall] Got event COMPLETED
Received response: com.apollographql.apollo.api.Response@cab19945
Second query value received

As you can see, even after cancelling the first call coroutine, the FoosCall is still active and even tries to reach out to the network (because the cache has been cleared, so behaving as expected as per apollographql/apollo-kotlin-normalized-cache-incubating#78)

I belive the bug is in the asFlow implementation which creates a clone() and subscribes to it, but it cancels the originally passed watcher, thus leaving the cloned one open and leaking. Simply capturing cloned watcher and cancelling it instead of this@toFlow fixes the issue

[tasomaniac]

Thanks for the detailed issue description. Your solutions sounds reasonable. Can you open a Pull Request with that fix? It would be awesome to include this to upcoming 2.0.0 release.

[lwasyl]

@tasomaniac Opened #2175