How do you make make a subscriber to a kotlin sharedflow run operations in parallel?

user_1771187 0
Viewed 507

I have a connection to a Bluetooth device that emits data every 250ms

In my viewmodel I wish to subscribe to said data , run some suspending code (which takes approximatelly 1000ms to run) and then present the result.

the following is a simple example of what I'm trying to do

Repository:

class Repo() : CoroutineScope {
    private val supervisor = SupervisorJob()
    override val coroutineContext: CoroutineContext = supervisor + Dispatchers.Default
    private val _dataFlow = MutableSharedFlow<Int>()
    private var dataJob: Job? = null
    val dataFlow: Flow<Int> = _dataFlow

    init {
         launch {
            var counter = 0
            while (true) {
                counter++
                Log.d("Repo", "emmitting $counter")
                _dataFlow.emit(counter)
                delay(250)
            }
        }
    }

}

the viewmodel

class VM(app:Application):AndroidViewModel(app) {
    private val _reading = MutableLiveData<String>()
    val latestReading :LiveData<String>() = _reading

    init {
        viewModelScope.launch(Dispatchers.Main) {
            repo.dataFlow
                .map {
                    validateData() //this is where some validation happens it is very fast
                }
                .flowOn(Dispatchers.Default)
                .forEach {
                    delay(1000) //this is to simulate the work that is done,
                }
                .flowOn(Dispatchers.IO)
                .map {
                   transformData() //this will transform the data to be human readable 
                }
                .flowOn(Dispatchers.Default)
                .collect {
                    _reading.postValue(it)
                }
        }

    }
}

as you can see, when data comes, first I validate it to make sure it is not corrupt (on Default dispatcher) then I perform some operation on it (saving and running a long algorithm that takes time on the IO dispatcher) then I change it so the application user can understand it (switching back to Default dispatcher) then I post it to mutable live data so if there is a subscriber from the ui layer they can see the current data (on the Main dispatcher)

I have two questions

a) If validateData fails how can I cancel the current emission and move on to the next one?

b) Is there a way for the dataFlow subscriber working on the viewModel to generate new threads so the delay parts can run in parallel?

the timeline right now looks like the first part, but I want it to run like the second one Execution timeline

Is there a way to do this?

I've tried using buffer() which as the documentation states "Buffers flow emissions via channel of a specified capacity and runs collector in a separate coroutine." but when I set it to BufferOverflow.SUSPEND I get the behaviour of the first part, and when I set it to BufferOverflow.DROP_OLDEST or BufferOverflow.DORP_LATEST I loose emissions

I have also tried using .conflate() like so:

repo.dataFlow
    .conflate()
    .map { ....

and even though the emissions start one after the other, the part with the delay still waits for the previous one to finish before starting the next one

when I use .flowOn(Dispatchers.Default) for that part , I loose emissions, and when I use .flowOn(Dispatchers.IO) or something like Executors.newFixedThreadPool(4).asCoroutineDispatcher() they always wait for the previous one to finish before starting a new one

Edit 2:

After about 3 hours of experiments this seems to work

 viewModelScope.launch(Dispatchers.Default) {
        repo.dataFlow
            .map {
                validateData(it)
            }
            .flowOn(Dispatchers.Default)
            .map {
                async {
                    delay(1000)
                    it
                }
            }
            .flowOn(Dispatchers.IO) // NOTE (A)
            .map {
                val result = it.await()
                transformData(result)
            }
            .flowOn(Dispatchers.Default)
            .collect {
                _readings.postValue(it)
            }
    }

however I still haven't figured out how to cancel the emission if validatedata fails

and for some reason it only works if I use Dispatchers.IO , Executors.newFixedThreadPool(20).asCoroutineDispatcher() and Dispatchers.Unconfined where I put note (A), Dispatchers.Main does not seem to work (which I expected) but Dispatchers.Default also does not seem to work and I don't know why

kotlin-sharedflow kotlin-coroutines kotlin android
1 Answers

First question: Well you cannot recover from an exception in a sense of continuing the collection of the flow, as per docs "Flow collection can complete with an exception when an emitter or code inside the operators throw an exception." therefore once an exception has been thrown the collection is completed (exceptionally) you can however handle the exception by either wrapping your collection inside try/catch block or using the catch() operator.

Second question: You cannot, while the producer (emitting side) can be made concurrent by using the buffer() operator, collection is always sequential.

As per your diagram, you need fan out (one producer, many consumers), you cannot achieve that with flows. Flows are cold, each time you collect from them, they start emitting from the beginning.

Fan out can be achieved using channels, where you can have one coroutine producing values and many coroutines that consume those values.

Edit: Oh you meant the validation failed not the function itself, in that case you can use the filter() operator.

The BroadcastChannel and ConflatedBroadcastChannel are getting deprecated. SharedFlow cannot help you in your use case, as they emit values in a broadcast fashion, meaning producer waits until all consumers consume each value before producing the next one. That is still sequential, you need parallelism. You can achieve it using the produce() channel builder.

A simple example:

val scope = CoroutineScope(Job() + Dispatchers.IO)

val producer: ReceiveChannel<Int> = scope.produce {
    var counter = 0
    val startTime = System.currentTimeMillis()
    while (isActive) {
        counter++
        send(counter)
        println("producer produced $counter at ${System.currentTimeMillis() - startTime} ms from the beginning")
        delay(250)
    }
}

val consumerOne = scope.launch {
    val startTime = System.currentTimeMillis()
    for (x in producer) {
        println("consumerOne consumd $x at ${System.currentTimeMillis() - startTime}ms from the beginning.")
        delay(1000)
    }
}

val consumerTwo = scope.launch {
    val startTime = System.currentTimeMillis()
    for (x in producer) {
        println("consumerTwo consumd $x at ${System.currentTimeMillis() - startTime}ms from the beginning.")
        delay(1000)
    }
}

val consumerThree = scope.launch {
    val startTime = System.currentTimeMillis()
    for (x in producer) {
        println("consumerThree consumd $x at ${System.currentTimeMillis() - startTime}ms from the beginning.")
        delay(1000)
    }
}

Observe production and consumption times.

user_108109630
Related