Wondering, if I have a Flow and I need that if there's an exception that emission should be skipped, and the next upstream emission should be processed, how do I do that?

Hello, does ThreadLocal works with coroutines?

This may be a dumb question but is coroutines flow still in beta?

hello, I'm trying to figure out the proper way to do interop between reactor and coroutines - I need to populate reactor subscriber context with value computed from coroutine.

Is there a

*.also {}

function that takes a suspending block of code?

scope.launch(<http://Dispatchers.IO|Dispatchers.IO>) {
   doSomeWork() // take 1 second
}

does using coroutine create memory footprint? because it has to use

Thread

at some point right? If I just launch 1 coroutine at the beginning and my app stay alive for 3 hours, does that

Thread

stay in memory for 3 hours?

What's the

Flow

equivalent to Rx Merge ? http://reactivex.io/documentation/operators/merge.html

How would I model this in flow? I have a

Flow<Int>

. If it doesn't emit anything within 200ms, emit

0

?

Hello, quick question about suspend and cancellation, what would be the best way to cancel early a suspend method that iterate on a list (using map operator) ?

suspend fun mappingFunction(dataToMap: List<String>) {
        val mappedData = dataToMap.map { 
            
        }
        
        // run suspendable work with mappedData
    }

Late night debugging before a release here - and

flatMapLatest

isn't behaving as we expect; expectation is it should be dropping the ongoing Flow when a new element is processed, like Rx

switchMap

Hey, what are the best practices to initialize a class

private

property with a suspending function return value (suspending supplier, pretty common when dealing with network)? Should I use

lateinit var

? Nullable

var

? Is something like

suspend init { }

planned at some point?

Is it an oversight that there’s no way to convert

Observable -> Flow

in the Flow RX extensions? You can do

Observable -> Flowable -> Flow

but it would be nice to be able to do convert it directly

I am new to coroutines. Lets say I have a suspend function that returns a nullable type. The type has a List field, whose values I read from a db using r2dbc, hence they are returned as a

Flow

. How do I map from a

Flow

to a nullable instance that my suspend function returns? In Reactor there are functions to map a

Flux

to a

Mono

..

Hello, can anyone answer me something about

kotlinx.coroutines.rx2

? I'm wondering if i use

rxSingle

, do i really need a context?

* Creates cold [single][Single] that will run a given [block] in a coroutine.
 * Every time the returned observable is subscribed, it starts a new coroutine.
 * Coroutine returns a single value. Unsubscribing cancels running coroutine.

Feels like as long i subscribes and unsubscribe in the right times it takes care of the coroutine and it doesn't cause memory leaks. Like this:

private fun single() = rxSingle {
    suspendedFun()
}

Is this correct to assume? I also can't provide a Context with a job to cancel it anyways.

2) Is it a compiler bug that suspending function references are no longer accepted if I remove the suspend modifier?

Hi, does anyone have had success converting from Rx2 Observable to Kotlin Flow, and getting data?

Hi, I see a lot of examples for attaching

CoroutineScope

to Android lifecycles with a pattern like that:

class UiLifecycleScope : CoroutineScope, LifecycleObserver {

    private lateinit var job: Job
    override val coroutineContext: CoroutineContext
        get() = job + Dispatchers.Main

    @OnLifecycleEvent(Lifecycle.Event.ON_START)
    fun onCreate() {
        job = Job()
    }

Do you guys see any reason for constructing the context with a getter, creating a new context every time. How about making the scope

var

?

override lateinit var coroutineContext: CoroutineContext
    @OnLifecycleEvent(Lifecycle.Event.ON_START)
    fun onCreate() {
        coroutineContext = Dispatchers.Main + Job()
    }

Question: Should not

runBlocking(Dispatchers.Main) {}

be the same as running

Handler(Looper.getMainLooper()).post{}

on android?

Is there a difference between

launch(newSingleThreadContext("ServiceContext")) {
 ...
}

and

launch {
 withContext((newSingleThreadContext("ServiceContext") {
    ....  
  }
}

?

coroutines on the jvm, is there some kind of underlying executor service that can be grabbed?

Hi everyone, trying out Channels for the first time. My use case is a “notification handler” type class where I can invoke

notify(...)

on an object of that class and it internally, with back-pressure awareness, processes and runs some computations on each notification. This is on Android, and for this example creating an

appScope

by using

GlobalScope + CoroutineExceptionHandler

because I want to treat the processing of the notifications as “fire-and-forget”. Also trying to use

consumeAsFlow()

+

collect()

on the Channel:

class ChannelExample {
    
    private val appScope = GlobalScope + CoroutineExceptionHandler { coroutineContext, throwable -> 
        // Log message + throwable
    }
    private val numbersChannel = Channel<Int>()

    init {
        appScope.launch(Dispatchers.Default) {
            numbersChannel
                .consumeAsFlow()
                .collect { number -> handleNumber(number)}
        }
    }
    
    fun notifyNumber(number: Int) {
        appScope.launch {
            numbersChannel.send(number)
        }
    }

    private suspend fun handleNumber(number: Int) {
        try {
            // Do some computations
        } catch (ce: CancellationException) {
            throw ce
        } catch (ex: Exception) {
            // Log message + exception
        }
    }
}

1) do I need to call

close()

or

cancel()

on

numbersChannel

? - if yes, where would be the best place to do that? - if no, why not? 2) are there any other things in this example that stray from best practices? 3) are there better ways to achieve my original use case? Thanks!

Then the call site looks super tidy:

fun getInvoice(id: String) {
        
        launchWithErrorHandling {

            val invoice = repository.getInvoice(id)
            _showSuccess.postValue(invoice)
        }
    }

🤔 interesting. So bottom line is better not mess with it.

What would be the Beh.Relay equivalent?

Hi, are actors to be used with flows? or are they different paradigm? Im trying to move my rx stuff to flows. translates easily. But I hear actors are nice in a way to bound paralelism (downloader queue) which Id find handy. But I see examples only actors receving and sendign to another actors, which leads me to believe thata its different abstraction alterantive to flows?

Hello. I have a flow created with

channelFlow

, and a simple flow created with

flowOf

. I want to concatenate them so I do

val concatened = chanFlow.onCompletion { if(it==null) emitAll(simpleFlow) }

. Then the flow is "launched"

GlobalScope.launch { concatened.collect { ... } }

. When I cancel the resulting job, the chanFlow stop correctly and throw

CancellationException

, but the second flow values are still collected. The

throwable

in

onCompletion

is null. What is the correct way to handle this ?

What would be the easiest way to limit parallelism on IO dispatcher? I.e. without creating a separate Executor thread pool, reusing the IO threads, but imposing a logical limit in a given function?

since combineLatest with multiple sources is now deprecated, how to get the latest result from multiple flows ?

there are no extention functions to convert from Observable to flow right? We need to convert it to flowable, and then use

.asFlow

?