channel.send()

suspend (I)->O

List<O>

callbackFlow

close(exception)

callbackFlow {
   try {
      val result = somethingThatThrows() // Try-catch or just let it be?
      offer(result)
   } catch (exception) {
      close(exception)
   }
   
   awaitClose { ... } 
}

flow.count() == 0 ?

internal class MainScope<T>(private val coroutineContextProvider: CoroutineContextProvider, private val response: (Response<T>) -> Unit) : CoroutineScope {
    override val coroutineContext: CoroutineContext
        get() = coroutineContextProvider.context() + job + exceptionHandler

    private val job = SupervisorJob()

    //This needs more investigation with thread scope
    private val exceptionHandler = CoroutineExceptionHandler { _, throwable ->
        logWarn(throwable.message.orEmpty())
        response.invoke(Response(error = throwable, value = null))
    }
}

class Response<out T>(val value:T?, val error: Throwable? = null) {

    fun isSuccess(block: (T) -> Unit): Response<T> {
        value?.let(block)
        return this
    }

    fun isFailure(block: (Throwable) -> Unit): Response<T> {
        error?.let(block)
        return this
    }

}

MainScope(coroutineContextProvider, onResponse).launch { 
 //CODE
}

java.lang.VerifyError: Verifier rejected

4.0.0-rc01

3.6.3

Foo

Bar

Bar(Foo)

Foo

Bar(Foo)

Foo

fun readAll(): Stream<Metadata> {
    return runBlocking {
        channelFlow<Stream<Metadata>> {
            Type.supportedTypes.forEach { type ->
                launch(<http://Dispatchers.IO|Dispatchers.IO> + CoroutineName("Read all for $type")) {
                    send(metadataPersistenceHelper(type).readAll())
                }
            }
        }.reduce { accumulator, value -> Stream.concat(accumulator, value) }
    }
}

Type

metadataPersistenceHelper

fun concat(first: Flow<Int>, second: Flow<Int>): Flow<Int> = flow {
    first.collect { value ->
        return@collect emit(value)
    }

    second.collect { value ->
        return@collect emit(value)
    }
}

return@collect

async { suspendFunc() }.await()

suspendFunc()

Spent the last 2 hours trying to figure out why this won't work (won't print anything) when I have newFixedThreadPoolContext(1) but works if I have 2 or more threads.  Can someone explain why that is. Maybe I am not using callbackFlow correctly?


 
   @Test
    fun `read range`() {
        val ctx = newFixedThreadPoolContext( 1, "tThread") /// test only works if 2 or more
        val scope = CoroutineScope(ctx)
        runBlocking {
            repeat(500) {
                repo.save("test$it")
            }
            scope.launch {
                repo.readRange().collect {
                    println(Belief.fromBytes(it))
                }
            }.join()
        }
    }
    
        fun readRange(): Flow<ByteArray> {
        return callbackFlow {
            database.readAsync { tr ->
               // reference: <https://apple.github.io/foundationdb/javadoc/com/apple/foundationdb/async/AsyncIterable.html>
               // will interate all values in the database
                val itr = tr.getRange(path.range()).iterator()

                // hasNext() is blocking (will do network call and bring back a few items)
                while (itr.hasNext()) { 
                    sendBlocking(itr.next().value)
                }
                channel.close()
                CompletableFuture.completedFuture(Unit)
            }
            awaitClose { }
        } // end of callbackFlow
    }

GlobalScope

GlovalScope

kotlinx.coroutines.JobCancellationException: StandaloneCoroutine was cancelled;

Promise

kotlinx.coroutines

.send

.send

.close

.send

.consume

.consumeEach