can anybody using coroutine 1.4.0-M1 confirm that a

MutableSharedFlow(replay = 0)

doesn't emit values ? 😕 the same code works if replay = 1 (but I don't want to replay first value each time 😢 )

Hello on my presenter I'm getting a

CoroutineScope

and a

CoroutineDispatcher

is it mandatory to implement in the

onDestroy()

the cancel of them? what happens if I don't do this?

Hi all! I'm trying to call 2 consecutive `Flow`s from an android

Activity

, I'm getting:

java.lang.IllegalStateException: ReceiveChannel.consumeAsFlow can be collected just once

The code I'm executing from is this:

private fun getCustomer() = lifecycleScope.launch {
        isLoading = true

        // Get the user's customer code.
        GetUserDetailsUseCase().getCustomer().collect { result ->
            when (result) {
                is Result.Success -> uploadDocument(result.data)
                is Result.Error -> showPopup(result.errorMessageResId)
            }
        }
    }

    private fun uploadDocument(customer: Customer) = lifecycleScope.launch {
        this@SignUpDocumentValidationActivity.customer = customer
        // Upload document and get URL.
        val filename = "${System.currentTimeMillis()}${customer.code}"
        when (val uploadResult = UploadImageUseCase().uploadImage(
            scaleImage(),
            filename
        )) {
            is Result.Success -> {
                Logger.i(TAG, "Image $filename.jpg uploaded successfully.")
                sendDocumentAndContinue(uploadResult.data)
            }
            is Result.Error -> showPopup(uploadResult.errorMessageResId)
        }
    }

    private fun sendDocumentAndContinue(documentUrl: String) = lifecycleScope.launch {
        // Send document to backend
        UpdateDocumentUseCase().updateDocument(
            customer!!,
            DocumentType.values()[currentStep],
            documentUrl
        ).collect { result ->
            when (result) {
                is Result.Success -> {
                    deleteImage()
                    isLoading = false
                    nextStep()
                }
                is Result.Error -> showPopup(result.errorMessageResId)
            }
        }
    }

How can I get rid of this issue and call the two collect() methods secuentially? Thanks!

Is it possible to use coroutine code from Java?

I have a

suspend

function that launches several operations (syncing individual pieces of local data) and I don't want this function to be run more than once simultaneously (if we try to POST the same data twice, then we get two new entries on the server side). What is the preferred practice for preventing a second simultaneous execution of this function? I have an idea, but I wanted the community opinion since there was some concern about my answer in a code review.

Is a

BroadcastChannel(BUFFERED)

still the best replacement for PublishSubject?

looking for a flow operator that does something like input: Foo output: List<Foo> everytime the upstream flow emits, the operator would emit a list of this value and any previous values that are emitted

I have seen examples of using GlobalScope.launch and runBlocking to create coroutines. How i can create coroutines without using the above two options? Using runBlocking as coroutine builder launches coroutines on a single thread, but i don't want to that. And GlobalScope launches in the root scope and has to be taken care of. What other options i have?

Hello guys, I have a function

buildRequestUrl

which build the URL including headers and query parameters. Now I am working on tests for api call's response. and I am debugging the code where I build the URL which is this

val uriBuilder = Uri.parse(baseUrl).buildUpon()

But the issue is it throws null pointer exception and result is null (see attached screenshot) How do I solve this issue? I am using below library

testImplementation 'junit:junit:4.13'
    testImplementation "org.mockito:mockito-core:3.4.6"
    testImplementation 'androidx.arch.core:core-testing:2.1.0'
    testImplementation 'org.jetbrains.kotlinx:kotlinx-coroutines-test:1.3.9'
    androidTestImplementation 'androidx.test.ext:junit:1.1.1'

I am using below class

@ExperimentalCoroutinesApi
class TestCoroutineRule : TestRule {

    private val testCoroutineDispatcher = TestCoroutineDispatcher()

    private val testCoroutineScope = TestCoroutineScope(testCoroutineDispatcher)

    override fun apply(base: Statement, description: Description?) = object : Statement() {
        @Throws(Throwable::class)
        override fun evaluate() {
            Dispatchers.setMain(testCoroutineDispatcher)

            base.evaluate()

            Dispatchers.resetMain()
            testCoroutineScope.cleanupTestCoroutines()
        }
    }

    fun runBlockingTest(block: suspend TestCoroutineScope.() -> Unit) =
        testCoroutineScope.runBlockingTest { block() }
}

I am using the below rule in the test class:

@get:Rule
    val testInstantTaskExecutorRule: TestRule = InstantTaskExecutorRule()

    @get:Rule
    val testCoroutineRule = TestCoroutineRule()

I'm getting random

kotlinx.coroutines.TimeoutCancellationException: Timed out waiting for 15000 ms

(Not serring timeouts anywhere explicitly). Does anyone know how I could increase this?

I want to use

Flow.combine(Flow)

However, the second flow is not known until at some point in the progression of the original Flow (at which point, the second flow can be derived). At the moment, I achieve this by (and it feels like a very nasty hack) having a

MutableStateFlow

and updating that flow from a coroutine launched from within

onEach

(just before

combine

). So then we use

combine(MutableStateFlow)

. But how to do this properly?

I have a

Channel(Channel.UNLIMITED)

in use to act as a queue between coroutines but am having trouble with my receiving coroutine not seeing any of what the producer is giving. could there be an issue from performing both of this in the scope of

<http://Dispatchers.IO|Dispatchers.IO>

(which i believe is multithreaded in java) ? should i be using a single threaded coroutine scope / dispatcher for this to work?

I am not sure what i am doing wrong here. Can anybody point out the obvious. 😅

import kotlinx.coroutines.GlobalScope
import kotlinx.coroutines.launch
import kotlinx.coroutines.*
import kotlin.system.measureTimeMillis
import kotlinx.coroutines.delay

fun main() = runBlocking {
    println("${Thread.activeCount()} threads active at the start")
    val time = measureTimeMillis {
        createCoroutines(10_000)
    }
    println("${Thread.activeCount()} threads active at the end")
    println("Took $time ms")
}

suspend fun createCoroutines(amount: Int) {
    val jobs = ArrayList<Job>()
    for (i in 1..amount) {
        jobs += launch {
            delay(1000)
        }
    }
    jobs.forEach {
        it.join()
    }
}

We’re getting a flow of rows from the database (through r2dbc) like this:

db.execute { ... }.asType<Foo>().fetch().flow()

Now we want to return a Map. Two ways of doing that are:

db.execute { ... }.asType<Foo>().fetch().flow().fold(mapOf()) { acc, it -> acc + (transform(it.bar) to it.baz) }

or:

db.execute { ... }.asType<Foo>().fetch().flow().map(transform(it.bar) to it.baz).toList().toMap()

Which is better and why?

How do you get root coroutine context? Or can you walk through parent contexts?

At Airbnb, we use Kotlin and coroutines in some of our server-side code. The workload for each request is fully async — there’s no blocking code that runs during the life of the request. For the most part, things work just fine, as you’d expect. However, as we load test these services and bring them to a breaking point, we see latency sky rocket, queuing start to occur at the web server layer (we use Dropwizard and non-blocking jetty APIs, so under normal operation, there is absolutely no queuing), and eventually heap exhaustion as all of this work is queued and requests never complete. Now of course, some of this behavior is expected — every service has its breaking point 🙂. In our Java services, however, which has a more mature service framework, we use standard patterns such as CoDel (controlled delay, https://en.wikipedia.org/wiki/CoDel), adaptive queuing, etc, to perform some load shedding to ensure that, even under heavy load, while the service may degrade, it never gets into an unrecoverable state. This has worked well for us, and I’d like to do similar things in our Kotlin services that use Coroutines. Does anyone have any insight into how to apply a load shedding algorithm (bulkheading, controlled delay, etc) for coroutines, while still leveraging the default work-stealing dispatcher?

CoroutineScheduler

doesn’t expose any public APIs to get it’s current “state”, meaning that there’s no way to really get much insight into the state of the default dispatcher. TL;DR — I want to figure out how to structure my application such that, when we’re saturating the CPU (and thus all the default dispatcher threads are super busy doing “stuff”), I can start to throw some back pressure and reject some requests, rather than continuing to accept hundreds/thousands of requests and causing the server to completely become unresponsive. And notably, I want to be able to cause this behavior with a mechanism that doesn’t require a ton of tuning per-service — I don’t want to do this strictly with a “request rate limit” or something like that. I’d like it to be much more adaptive, ideally.

Where would you put a register/signUp method that needs to execute some extra asynchronous work after the signUp process itself is finished? When I put it into a SignUpViewModel that is scoped to SignUpFragment the viewmodelScope is canceled too early (because as soon as we signed up, we leave this fragment).

Anyone have an example of a producer and multiple worker consumer pattern using flow, shareIn?

anyone know why the coroutine debugger wouldn't be showing up in intellij / android studio? i'm using kotlin 1.4.10 and coroutines 1.3.9

Is

offset

variable in this method safe to use in multiple threads?

suspend fun InputStream.readByChunks(
    fileSize: Long,
    chunkSize: Int,
    block: suspend (bytes: ByteArray, offset: Long) -> Unit
) {
    var offset = 0L

    this.buffered().use { input ->
        while (offset != fileSize) {
            val buffer = if (offset + chunkSize < fileSize) {
                ByteArray(chunkSize)
            } else {
                ByteArray((fileSize - offset).toInt())
            }

            val read = input.read(buffer)
            block(buffer, offset)
            offset += read
        }
    }
}

has anyone else had trouble with using a channel in a multiplatform context? right now i'm testing on android but for some reason my iterator over my channel is not receiving any values. when i attempt to reduce the problem and run it in a jvm only project it works fine though

class PersistentWsClient() {
	private val scope = CoroutineScope(Dispatchers.Default)
    private val sendingQueue = Channel<Frame>(UNLIMITED)

    fun open(host: String, path: String, port: Int = 8080) {
        scope.launch {
            this@PersistentWsClient.e("waiting for messages")
            for (msg in sendingQueue) {
                this@PersistentWsClient.e(msg.toString())
            }
        }
    }

    fun send(payload: Frame) {
        scope.launch {
            for (i in 0..10) {
                delay(500)
                sendingQueue.offer(payload)
                this.e("offering $payload")
            }
        }
        sendingQueue.offer(payload)
    }
}

(i have stripped away some extra code). In this example i only ever get

waiting for messages
offering <payload>
offering <payload>
...

1. Why am I seeing: java.lang.NoClassDefFoundError: Could not initialize class kotlinx.coroutines.CancelledContinuation 2. On my local machine I don't get stack overflow, on our CI pipeline I get the

StackOverflowError

error

or other words, this of the lambda is Transaction instance and all consumer methods should be a extension on Transcation is it possible to enforce this without a compiler plugin?

2) coming from rxjava

Observable.just(1).startWith(0)

emits 0,1 flow

flowOf(1).onStart { emit(0) }

emits 1,0 is this by design? a rx fix? pretty foot gunny when mkgrating also, what if the actual flow is replayed? if its "below" the onStart, id assume it emits first which is not good

i have a suspend method that i want to make use of

suspend fun lookupListings(): List<QueryListings.SelectListingsResult> {

if i make use of runBlocking, i'm assuming it's physically blocking a thread until that suspend function completes, what is the better route to take here to get rid of the `runBlocking`s?

class HomePage : PageResponse(
	title = "Page Title",
	description = "Page Description",
) {

	val listings = runBlocking {
		lookupListings()
	}

    val somethingElse = runBlocking {
        suspendLookupSomethingElse()
    }

Edit: this is on the JVM

StateFlow

appears to compare values before updating them (unlike

LiveData

), so if I have a mutable list which I update and send to my

MutableStateFlow

it doesn't get published from the looks of it - unless I create a new list / copy it. Is that the right way to work around this?

Android android studio folks? This happens to me all the time

3) Its me with rxjava

startWith

and flow

onStart

again

CoroutineScope(SupervisorJob() + <http://Dispatchers.IO|Dispatchers.IO>).launch {
	flowOf(1)
		.onStart { emit(0) }
		.collect {
			stateFlow.value = it
		}
}

-- is equivalent to:
Observable.just(1)
	.startWith(0) <---
	.subscribeOn(<http://Schedulers.io|Schedulers.io>()) <---
	.subscribe {
		behaviorRelay.accept(it)
	}

-- In rx I use:
Observable.just(1)
	.subscribeOn(<http://Schedulers.io|Schedulers.io>()) <---
	.startWith(0) <--- swapped
	.subscribe {
		behaviorRelay.accept(it)
	}

is this doable somehow in flow? (to not have the onStart emit coming from io thread)

When declaring a suspend function within another suspend function. Is this a compiler bug?