CODEWITHSUNDEEP
javakotlinrx-java2
Uniqus
Uniqus

Reputation: 574

rxjava 2: how to dispose a resource after downstream completes processing

I have a (Flowable) stream of items to be processed in parallel using a single common resource, and the resource must be disposed afterwards. I tried to use Single.using() operator, but it disposes the resource before even the first item in the stream is processed.

Sample program (in Kotlin):

package my.test.rx_task_queue

import io.reactivex.Flowable
import io.reactivex.Single
import io.reactivex.schedulers.Schedulers
import org.slf4j.LoggerFactory
import java.util.concurrent.atomic.AtomicInteger

object TestCommonResource {
    private val logger = LoggerFactory.getLogger(TestCommonResource::class.java)
    @JvmStatic
    fun main(args: Array<String>) {
        val queue = Flowable.fromIterable(1..5)
        val resIdx = AtomicInteger(0)
        val resource = Single.using({
            val res = "resource-${resIdx.incrementAndGet()}"
            logger.info("Init resource $res")
            res
        }, { res ->
            Single.just(res)
        }, { res ->
            logger.info("Dispose resource $res")
        }, false)

        val result = resource.flatMap { res ->
            queue.flatMapSingle({ item ->
                Single.fromCallable {
                    logger.info("Process $item with $res")
                    "$item @ $res"
                }
                        .subscribeOn(Schedulers.io())
            }, false, 2)
                    .toList()
        }
                .blockingGet()
        logger.info("Result: $result")
    }
}

Sample log output:

14:30:27.721 [main] INFO my.test.rx_task_queue.TestCommonResource - Init resource resource-1
14:30:27.744 [main] INFO my.test.rx_task_queue.TestCommonResource - Dispose resource resource-1
14:30:27.747 [RxCachedThreadScheduler-1] INFO my.test.rx_task_queue.TestCommonResource - Process 1 with resource-1
14:30:27.747 [RxCachedThreadScheduler-2] INFO my.test.rx_task_queue.TestCommonResource - Process 2 with resource-1
14:30:27.748 [RxCachedThreadScheduler-3] INFO my.test.rx_task_queue.TestCommonResource - Process 3 with resource-1
14:30:27.749 [RxCachedThreadScheduler-4] INFO my.test.rx_task_queue.TestCommonResource - Process 4 with resource-1
14:30:27.749 [RxCachedThreadScheduler-1] INFO my.test.rx_task_queue.TestCommonResource - Process 5 with resource-1
14:30:27.750 [main] INFO my.test.rx_task_queue.TestCommonResource - Result: [1 @ resource-1, 2 @ resource-1, 3 @ resource-1, 4 @ resource-1, 5 @ resource-1]

Using Flowable.parallel() instead of flatMap() leads to the same result.

Upvotes: 0

Views: 628

Answers (1)

tynn
tynn

Reputation: 39873

The disposing happens with the disposal of the source, so if you want to dispose after everything is done, you just need have singleFunction return the whole stream:

object TestCommonResource {
    private val logger = LoggerFactory.getLogger(TestCommonResource::class.java)
    @JvmStatic
    fun main(args: Array<String>) {
        val queue = Flowable.fromIterable(1..5)
        val resIdx = AtomicInteger(0)
        val result = Single.using({
            val res = "resource-${resIdx.incrementAndGet()}"
            logger.info("Init resource $res")
            res
        }, { res ->
            queue.flatMapSingle({ item ->
                Single.fromCallable {
                    logger.info("Process $item with $res")
                    "$item @ $res"
                }
                        .subscribeOn(Schedulers.io())
            }, false, 2)
                    .toList()
        }, { res ->
            logger.info("Dispose resource $res")
        }, false)
                .blockingGet()
        logger.info("Result: $result")
    }
}

Upvotes: 1

Related Questions