Is it on purpose that

Flow

has

singleOrNull

but not

firstOrNull

?

I'm having issues testing expected exceptions in coroutines on Android. Full MWE in the attached code snippet. As far as I understand is that when a new coroutine is launched from a unit test, even if that indirectly happens on the

TestCoroutineDispatcher

or

TestCoroutineScope

, then any exceptions are swallowed (although the stack traces are printed). What is the general pattern to work with expected exceptions in unit tests and coroutines on Android?

Hi, anyone who could help me with

'More than one file was found with OS independent path 'META-INF/kotlinx-coroutines-core.kotlin_module'

?

How do I return a value from a flow? Trying to return the myString value

suspend fun doSomething(): String {
   val myFlow = flow { emit("hello") }
   myFlow.collect { myString ->
        return myString
   }
}

Hi guys, I’m still pretty new to Coroutines and was wondering if anyone can let me know if I’m on the right track for what I want. Problem: I’m currently building out an Android application and I have several screens that require the User’s Location. This application can have multiple Activities/Fragments in its backstack. If the user changes their location on a screen, I would want every screen in the backstack to get that updated location object as well. Question: Looking into

Flows

, is a

channelFlow

what I need for my particular use-case? There would be one singular source-of-truth that multiple call sites will be subscribing to for the User’s location and would be a cold stream as the location would only be updated based on a method invocation.

Hello, I am trying to concat two flows in a way that I get results sequentially when I collect on it

flowOf(
 remoteDataSource.getDataFromCache() // suspending function returning Flow<Data>
  .catch { error -> Timber.e(error) },
 remoteDataSource.getDataFromServer() // suspending function returning Flow<Data>
).flattenConcat().collect {
Timber.i("Response Received")
}

my expectation here was that I will get first result first and then second after sometime as response comes from server. But the problem here is collect only gets called after server returns result.

Hello, I'm playing with a custom

CoroutineDispatcher

implementation using this implementation as a reference: https://github.com/Kotlin/kotlinx.coroutines/blob/master/reactive/kotlinx-coroutines-rx2/src/RxScheduler.kt However

Delay

interface is marked as

InternalCoroutinesApi

. Is there a good way to implement delaying?

I am doing my first steps with coroutines. I wrote a simple multi-client echo server using ktor, can someone please review my code and point out potential spaghetti?

import io.ktor.network.selector.ActorSelectorManager
import io.ktor.network.sockets.*
import kotlinx.coroutines.*
import <http://kotlinx.coroutines.io|kotlinx.coroutines.io>.*
import <http://java.net|java.net>.InetSocketAddress
import java.nio.channels.ClosedChannelException

suspend fun main() {
	val echoServer = EchoServer()

	supervisorScope {
		launch {
			echoServer.start()
		}

		while (readLine() != "exit") {
			println("unknown command")
		}
		println("stopping server")
		echoServer.stop()
	}
}

class EchoServer {
	private val listener: ServerSocket =
		aSocket(ActorSelectorManager(<http://Dispatchers.IO|Dispatchers.IO>)).tcp().bind(InetSocketAddress(12345))
	private val clients = mutableListOf<Socket>()

	suspend fun start() {
		coroutineScope {
			while (true) {
				val client =
					try {
						listener.accept()
					} catch (_: ClosedChannelException) {
						break
					}
				launch {
					handleClient(client)
				}
			}
		}
	}

	private suspend fun handleClient(client: Socket) {
		println("new client connected")
		clients.add(client)
		val reader = client.openReadChannel()
		val writer = client.openWriteChannel(true)
		while (true) {
			val message = reader.readUTF8Line() ?: run {
				println("client disconnected")
				return
			}
			writer.writeStringUtf8("$message\n")
		}
	}

	fun stop() {
		listener.close()
		clients.forEach {
			it.close()
		}
		clients.clear()
	}
}

Is the order preserved in a Flow? Is there a way to tell it to run lambdas (

map

, etc) in parallel and send elements as soon as possible, even if the order is lost?

Flow question. the

sample

operation is almost exactly what I want. Except for that small warning

Note that the latest element is not emitted if it does not fit into the sampling window.

I want to collapse elements(just like sample does), but the last element at any time is kinda important. Any pointers?

what is the difference between the

block: suspend () -> T

and

block: suspend CoroutineScope.() -> T

types?

I need help with since Coroutine never calls API after job either cancelled or pending. Details: https://stackoverflow.com/questions/60216513/coroutine-never-calls-api-after-job-either-cancelled-or-pending

What is the best approach of starting a long running computation inside a

CoroutineScope

that keeps on running even if the scope is cancelled and if another (new) scope is accessing the same function/computation, the already computed value should be returned? Sounds like a job for a

ConflatedChannel

, right? I tried something with

Deferred

but the code (test code!!) is far from beautiful:

var result: Deferred<Int>? = null

fun longComputationAsync(): Deferred<Int> {
    result = result ?: GlobalScope.async {
        println("longComputation")
        delay(1000)
        1337
    }
    return result!!
}

val scope1 = CoroutineScope(Dispatchers.Default)
scope1.launch {
    println("scope 1")
    println(longComputationAsync().await())
}
scope1.cancel()

val scope2 = CoroutineScope(Dispatchers.Default)
val job = scope2.launch {
    println("scope 2")
    println(longComputationAsync().await())
}

runBlocking {
    job.join()
}

so basically coroutinecontext have a scope and these are coroutinescope which provides lifecycle to coroutine.

message has been deleted

I'm searching for a data structure that allows a consumer to request for some data one by one (eg. A function

next()

or something similar), and from one-to-many producers to add new data to it. That sounds awfully similar to what a Channel is, however from my understanding the consumer·s of a channel act as soon as an element is produced, rather than when they want to? Basically I'm searching for a way to have the client decide when new elements are produced, and that data structure would have some kind of buffer of size

n

so the client could almost always get the data instantaneously, and the producer·s would get to work immediately when the buffer gets too small to add new elements.

What's the equivalent of Rx's

PublishProcessor

in the Coroutines world?

Hi All, it might seem trivial, but I wanted to ask, shall cancel CoroutineScope in ViewModel$onCleared or shall I cancel in Activity$onDestroy. What is recommended???

Hey. I have an Rx library I use to connect to a BLE device. This library has this function:

fun connect(bluetoothDevice: BluetoothDevice) : Observable<RxBleConnection>

. The reason this returns an

Observable

instead of a

Single

is that the life-cycle of the

Observable

is tied to the connection. So this observable will only ever emit one item, but it will stay alive without completing so the connection stays alive until we dispose. I want to wrap this to use coroutines instead. I'm thinking about having a function like

suspend fun connect(bluetoothDevice : BluetoothDevice) : RxBleConnection

, Any ideas on how to handle the life-cycle of the Observable? If I just use

connectObservable.awaitFirst()

it will take the first value, and dispose, which will stop the connection.

unless you change the dispatcher of emmit

will channelFlow or callbackFlow be reaching Stable anytime soon?

Is there a way to implement dynamic timeout for a coroutine? Basically if I'm running callable A and B in parallel with some timeout t, I want to extend the timeout of callable A based on the response from callable B (which should be faster). Would I just have to avoid using withTimeout and instead keep track of the time and cancel manually?

Should I use

withContext(<http://Dispatchers.IO|Dispatchers.IO>) { }

when using

ConcurrentHashMap

inside a suspending function?

Anyone have tips for installing java 1.6 on a mac to run tests from the coroutines repo?

How do you guys usually handle exceptions in coroutines while still propagating cancelations? It always feels a little clunky and with too much boiler plate since coroutines are built to use exceptions but kotlin doesnt have checked exceptions so you usually have to do

try {
    exceptionalMethod()
catch (e: Exception) {
    if (e is CancellationException) {
       throw e
    }
    // handle other exceptions 
}

Would it be worth writing a some sort of top level function similar to runCatching that does the above? Or are there cleaner solutions?

how do I switch between dispatchers using

async

and

await

?

Did something change with iosArm32 for

Expected property 'PLATFORM_DISPATCHER' has no actual declaration in module 
The following declaration is incompatible because return type is different:
    internal actual var PLATFORM_DISPATCHER: [ERROR : CoroutineDispatcher]

I can't compile my iosArm32 module after updating to 1.3.61

Could anyone explain simply what the responsiblity of the

continuation

object which is passed around with suspend functions?

How do I print all the handler names with below program? Currently, it's only printing 0th handler name and waiting for it to finish:

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.async
import kotlinx.coroutines.runBlocking
import java.util.stream.IntStream

fun main() {
    runBlocking {
        val asyncs = IntStream.range(0, 6).mapToObj {
            GlobalScope.async {
                handleData("handler-$it")
            }
        }
        asyncs.forEach {
            runBlocking {
                it.await()
            }
        }
    }
}

private fun handleData(handlerName: String) {
    println(handlerName)
    while (true) {
    }
}

Current output: handler-0 Expected output: handler-0 handler-1 handler-2 handler-3 handler-4 handler-5