Easy Unsubscription

[ZakTaccardi]

One of the best parts of Rx is that unsubscriptions automatically transfer upstream. I can create a complex stream from multiple data sources, transform them, and subscribe to them. Then, I can add them to a CompositeDisposable - which can automatically get cleared during the appropriate lifecycle event, like onDestroyed() (I'm an android developer). It would be really helpful if co-routines could implement this behavior.

val disposables = CompositeDisposable()

// writeable stream in our model layer
val writeableSource: Subject<Int> = PublishSubject.create()

// readable stream. our model layer only exposes readable properties
val readableSource: Observable<Int> = writeableSource

val disposable = readableSource.map {
    println("map executed for: $it")
    it
}
    .subscribe { println(it) }
    .addTo(disposables)

writeableSource.onNext(1)

disposables.clear()

// should be ignored, because we previously called clear()
writeableSource.onNext(2)

// outputs:
// map executed for: 1
// 1

The biggest roadblock to this is that because Channel<T> subscriptions are hot, unsubscriptions do not automatically transfer upstream. Would #254 provide this? More specifically:

// writeable stream in our model layer
val broadcastChannel = BroadcastChannel<Int>(10)

// readable stream
val source: SubscriptionReceiveChannel<Int> = broadcastChannel.openSubscription()

launch {
    source // closing this prevents downstream operators from emitting
        .map { it + 1 }
        .map { it + 2 }
        .also { receiveChannel ->
            // calling `.cancel(null)` on the `ReceiveChannel` returned by the previous
            // `.map` operator above does not close the upstream channel. This can easily
            // cause a memory leak
            receiveChannel.cancel(null)
        }
        // ideally, `.consumeEach()` could somehow be modified to return a `Disposable` like
        // interface that cancels all upstream emissions
        .consumeEach { println(it) }
}

[ZakTaccardi]

If this is exactly the type of behavior #254 is meant to provide, please close this in favor of it.

[jcornaz]

In your latest code example, if you cancel the job returned by launch, it would make the receive method performed by consumeEach to throw a cancellation exception. And, because consumeEach use consume, any exception thrown (including the cancellation exception) would cancel the channel upstream. And because the upstream is a subscription, cancelling means "unsubscribing".

So your code does return a kind of "disposable". It is returned by launch instead of consumeEach as you were expecting. and it is called Job not Disposable.

Aggregating the cancellation can easily be made with cancellation via explicit job.

For example:

val compositeDisposable = Job()

launch(parent = compositeDisposable) {
   channel.consumeEach { println(it) }
}

compositeDisposable.cancel() // This cancel all child jobs.

[jcornaz]

// calling .cancel(null) on the ReceiveChannel returned by the previous
// .map operator above does not close the upstream channel. This can easily
// cause a memory leak
receiveChannel.cancel(null)

This is not correct. cancelling the result of map, filter or any other operator built in "kotlinx.coroutines". does cancel the upstream channel. (If you build your own operator, you should take care of it. The easiest way, is to use produce and consume)

You can easily convince yourself if you print something when cancel is called:

fun main(args: Array<String>) {

    // writeable stream in the model layer
    val broadcastChannel = BroadcastChannel<Int>(10)

    // readable stream
    val sub = broadcastChannel.openSubscription()

    // use a delegate to print something when cancel is called on the subscription
    val source: SubscriptionReceiveChannel<Int> = object : SubscriptionReceiveChannel<Int> by sub { 
       override fun cancel(cause: Throwable?): Boolean {
            println("cancelling subscribtion with cause: $cause")
            return sub.cancel(cause)
        }
    }

    val job = launch {
        source // closing this prevents downstream operators from emitting
                .map { it + 1 }
                .map { it + 2 }

                // calling `.cancel(null)` on the `ReceiveChannel` returned by the previous
                // `.map` operator above DOES close the upstream channel. This easily
                // avoid memory leaks
                .cancel(null)
    }

    runBlocking { job.join() }
}

You'll see that this code does print "cancelling subscribtion with cause: null" because of the receiveChannel.cancel(null).

[elizarov]

@jcornaz is absolutely right that channel operators like filter are supposed to be inherently safe and are designed to cancel the upstream data source problem without any need to manually manage it with CompositeDisposable like you have to do with Rx.

Note, that current implementation of filter and others is slightly buggy and may fail to cancel upstream source under some very specific circumstances that are unlikely to happen in real-life code, but still see #279 for details.

I'm closing this particular issue anyway.

[ZakTaccardi]

without any need to manually manage it with CompositeDisposable like you have to do with Rx.

I'm confused, because there still is a need for multi-job cancellation. I would recommend adding a CompositeJobs/CompositeCloseable (or something similar) class whose purpose is to to cancel multiple in flight jobs.

Is there a reason ReceiveChannel<T> and Job do not implement Closeable?

[JakeWharton]

Pass a parent Job, cancel the parent.

Closeable is JVM-specific.