Hey, Im writing a adapter for android sharedpreferences listener into flow

private fun SharedPreferences.onSharedPreferenceChangedFlow(): Flow<String> {
    return callbackFlow {
        val listener = SharedPreferences.OnSharedPreferenceChangeListener { _, key ->
            trySend(key)
        }
        registerOnSharedPreferenceChangeListener(listener)
        awaitClose { unregisterOnSharedPreferenceChangeListener(listener) }
    }
}

seems to work, however I do know there is one nasty behavior of the

SharedPreferences.registerOnSharedPreferenceChangeListener

, that it only keeps weak references, so the listener might get gc-ed so I need to hard reference the listener How would I turn it into a field? Or rather, I think I need a proper CallbackFlow subclass, not just use the builder, right? And looking at sources, everything is private / internal api

Hey, how do you observe SharedFlow in Jetpack Compose? I am currently observing StateFlow in Compose like so: val state by myViewModel.myStateFlow.observeAsState() but was wondering what the right thing to do is with SharedFlow

Is Ktor doing it all wrong? Hoping to find some answers/opinions on this channel. https://kotlinlang.slack.com/archives/C0A974TJ9/p1640968234268400

I use the following code to create a fake implementation of a countdown timer based on Coroutines. Can someone tell me if this will be exact or if I need to calculate the time difference in some other way? (see thread)

Why would I not always use Dispatchers.Main.immediate (over the nonimmediate)?

I have a callback based API which I tried to convert it into a hot flow using callbackFlow builder. Would you mind suggesting how to write a unit test which ensures that the API is correctly converted to a hot cold flow? Here is my StackOverflow question: https://stackoverflow.com/questions/70576533

hi, i just read https://maxkim.eu/things-every-kotlin-developer-should-know-about-coroutines-part-2-coroutinescope which says that one shouldn't implement the

CoroutineScope

interface anymore. unfortunately it doesn't say anything about the WHY despite using the builder function would be more straightforward. why is it more straightforward?

Hello, Is there a mature example of creating a

java.lang.reflect.Proxy

for a Kotlin interface with

suspend

methods? (I would call the proxy methods only from Kotlin code.) Thanks.

Hi, I am writing an android instrumentation test to test one of my fragments. The fragment uses a (real) viewmodel and I have declared mocks for it to use. The issue is one of these mocks (a repository class) has a suspend function and I am not sure what to do in the test.

I have a object list that is processed using parallelstream. How can I migrate this to corroutine?

Is anyone here familiar with ktor raw sockets? I'm seeing a strange behavior when I employ ktor raw sockets at scale and I'm hoping someone might be able to help.

Hello everyone! I have a question about best practices when using coroutines in a Spring MVC application. We are stuck on a version 2.3 for now and don't have access to coroutine support in the controller. So I use runBlocking in my controller methods, and call suspending functions in my services. Ultimately this will lead to blocking calls to another service. Is it better to run all coroutines including the one created by the runBlocking call with Dispatchers.IO by specifying it in the controller or should I just have the coroutines that make the blocking calls specify the dispatcher? I'm worried about testability mainly. Thanks!

Is it possible to tweak channel settings in callback or channelflow? or is it hardcoded to

BUFFERED

and expected for consumers to use

buffer

operator?

It seems that runTest is somehow broken with shared flows:

private suspend fun testBody(scope: CoroutineScope) {
    val value = MutableStateFlow(1)
    val sharingJob = Job()
    val flow = value.shareIn(scope + sharingJob, SharingStarted.WhileSubscribed(replayExpirationMillis = 0), replay = 1)
    check(flow.first() == 1)
    value.value = 2
    check(flow.first() == 2)
    sharingJob.cancel()
  }

  @Test
  fun withRunBlockingTest() = runBlockingTest {
    testBody(this)
  }

  @Test
  fun withRunTest() = runTest {
    testBody(this)
  }

This passes with runBlockingTest but fails with runTest. It is possible to fix it by adding a runCurrent before calling flow.first a second time but I think thats not how it’s supposted to be and makes me very uncomfortable writing tests without having a runCurrent between every single line

How would you test this countdown timer based on coroutines? What can I use as the time source because I can't use

SystemClock.elapsedRealtime

in a test.

Is it possible to get hot flows or are all flows cold and I should rather use a channel? Use-case: logging When I have a 10k concurrent things happening and they are all logging I want to throw the log line on a "channel" and not block the call-site until the log is completed Then on a consumer side, I want to process the incoming logs in batches, grab a 100 at a time or until the "channel" is empty and do a single println for example or a single POST if it's going to an external logging server

In my following code, if

countDownInterval

is 0 I want to skip the loop and just delay until the time is over. The problem is that delay is inexact (that's why I do the comparison with the SystemClock time). How could I achieve a (somewhat) exact delay?

I'm wrapping java.nio.channels.AsynchronousFileChannel with a callbackFlow to convert an asynchronous File reader into a flow. Is there a better way of doing this or anything that I can improve on this implementation?

class FileFlow(private val uri: String) {

	private data class FileBaton(
		val fileChannel: AsynchronousFileChannel,
		val flow: ProducerScope<ByteArray>,
		val buffer: ByteBuffer,
		val position: Long,
		val handleError: (message: String, e: Throwable?) -> Unit = { message, e ->
			flow.cancel(message, e)
			runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
				fileChannel.close()
			}
		}
	)

	fun read(bufferSize: DataSize): Flow<ByteArray> = callbackFlow {

		val path = Path.of(uri)
		val baton = FileBaton(
			buffer = ByteBuffer.allocate(bufferSize.B),
			fileChannel = withContext(<http://Dispatchers.IO|Dispatchers.IO>) {
				AsynchronousFileChannel.open(path, StandardOpenOption.READ)
			},
			flow = this,
			position = 0,
		)

		baton.fileChannel.read(
			baton.buffer,
			baton.position,
			baton,
			object : CompletionHandler<Int, FileBaton> {
				override fun completed(read: Int, baton: FileBaton) {
					if (read > 0) {
						trySendBlocking(
							baton.buffer.array().sliceArray(0 until read)
						).onFailure { e ->
							baton.handleError(e?.message ?: "", e)
						}

						baton.buffer.rewind()
						baton.fileChannel.read(
							baton.buffer,
							baton.position + read,
							baton.copy(
								position = baton.position + read
							),
							this
						)
					} else {
						baton.flow.channel.close()
						runBlocking(<http://Dispatchers.IO|Dispatchers.IO>) {
							baton.fileChannel.close()
						}
					}
				}

				override fun failed(e: Throwable?, baton: FileBaton) {
					baton.handleError(e?.message ?: "", e)
				}
			}
		)
		awaitClose {

		}
	}
}

FileFlow("/home/.../blah.txt").read(250.kB).collect {
   print(String(it))
}

Hello, I’d like to invite some feedback on a design I have. I am trying to model a stateful connection using a

StateFlow<Job>

, so that connection lifecycles can be observed, cancelled, reconnected, etc. I have an Android app which connects to several websocket endpoints on different protocols, including plain websockets, socketio, action cable, graphql subscriptions, so being able to define a common pattern like this would be useful to me. More in 🧵

I've updated to 1.6.0-native-mt and on iOS it has broken a lot of my test cases, I found https://youtrack.jetbrains.com/issue/KTOR-3612 but I am also having issues for tests that cover code that calls stateIn. I am currently doing

.stateIn(
    CoroutineScope(Dispatchers.Default),
    SharingStarted.WhileSubscribed(),
    null
)

and getting

Cannot start an undispatched coroutine in another thread DefaultDispatcher from current MainThread

as an error from runTest

I'm trying to figure out how to best set up coroutines/flows to show updates to nested objects. Specifically, I need to display a

Group

object, which has some normal data to display (e.g.

id

), but also has a list of

User

objects (which each have their own data to display) I'd of course like my database to have group data separated from the

User

data, since `User`s can exist without a group, with multiple groups, etc. So, each group in the db will simply list the relevant

User

IDs; I call this a

ShallowGroup

I know how to get a flow to observe a single

User

, or how to get a flow to observe a single

ShallowGroup

but I can't figure out how to get from those a flow that produces full

Group

objects that reflect the changes to the inner

User

objects. My best understanding of how it would work: 1. Get a flow of

ShallowGroup

objects. 2. In a

map

call on that flow, use the user IDs in that

ShallowGroup

snapshot to get a flow for each

User

, and: 3. use

combine

on the previous list of flows of

Users

to get a flow which produces `List<User>`/`Group` 4. ?? somehow flatten the

Flow<Flow<Group>>

to just a

Flow<Group>

?? Note that the contents of each

ShallowGroup

object received determine which

User

objects need to be observed. Can someone point me to what the best way is to handle flows for nested objects like this?

Can anybody explain me why do sequenced setting value in MutableStateFlow doesn’t emit until last value set? there is simplified example of what i do in my app in thread. i try to collect combined value from state flow, and emits are sequenced (lines 25, 26)

Sorry if it was already mentioned here, this is a very interesting project: https://www.reddit.com/r/Kotlin/comments/rwswao/kotlin_coroutines_stack_trace_issue/

I want to register an observer to a lifecycleOwner and remove it when this lifecycle is no longer resumed. I am currently doing:

lifecycleOwner.repeatOnLifecycle(Lifecycle.State.RESUMED) {
    launch {
        val callback = LifecycleEventObserver { _: LifecycleOwner, event: Lifecycle.Event ->
            if (event == Lifecycle.Event.ON_PAUSE) {
                // do something on pause
            }
        }
        lifecycleOwner.lifecycle.addObserver(callback)
        try {
            awaitCancellation()
        } finally {
            lifecycleOwner.lifecycle.removeObserver(callback)
        }
    }
}

But this

try/finally

with the

awaitCancellation()

combination feels a bit weird. Is there a nicer way to do this in general?

I am writing a custom flow and inside it I want to non-blocking collect some other flow, process it's emissions and emit them:

val externalFlow = flowOf(1,3,4)
val myFlow = flow<Int> {
  coroutineScope { externalFlow.collect { this@flow.emit(it + 8) } }
  emit(1)
  emit(99)
}

I've used

coroutineScope

, but I'm worried that it won't be cancelled when

myFlow

collection is cancelled or interrupted. It's just that I've found internal

flowScope

builder in

coroutines.core

and it says:

This builder is similar to coroutineScope with the only exception that it ties lifecycle of children and itself regarding the cancellation, thus being cancelled when one of the children becomes cancelled.

Can I achieve similar effect for my case? Or using

coroutineScope

like I did is ok?

I’m trying to understand when I need to use

suspendCancellableCoroutine

over

suspendCoroutine

. If I got a use case where all I need to do is something like:

suspend fun MediaPlayer.seekToPercent(
    @FloatRange(from = 0.0, to = 1.0) percentage: Float,
) = suspendCoroutine<Unit> { cont ->
    val callback = MediaPlayer.OnSeekCompleteListener {
        this.setOnSeekCompleteListener(null)
        cont.resume(Unit)
    }
    setOnSeekCompleteListener(callback)
    val positionToSeekTo = (duration.toFloat() * percentage).toInt()
    seekTo(positionToSeekTo)
}

Is there just no reason to use

suspendCancellableCoroutine

? Basically I wonder if there are any other implications regarding the cancellability of

suspendCoroutine

and if it behaves in any different way to

suspendCancellableCoroutine

, or all that they do differently is that in a

suspendCancellableCoroutine

we have access to more than just

resume(value: T)

and

resumeWithException(exception: Throwable)

? Reading the documentation doesn’t fill all the gaps in my head

Kinda similar question as above, but maybe a little simpler. I'm reading the documentation for Flows, and I'd like to implement a Flow that executes each of the items concurrently. And when I'm reading the docs, I see https://kotlinlang.org/docs/flow.html#buffering which explains quite well that this should be possible without explicitly calling

launch

or

async

on the calling function. I'm trying to implement this as close to the example as I can, but no matter what I do, it doesn't seem to be executing the items in the flow concurrently, but only sequentially. Here's a simplified code snippet of what I'm attempting:

enum class Ordinal(val delay: Long) {
		FIRST(100),
		SECOND(200),
		THIRD(300),
		FOURTH(400),
		FIFTH(500),
		SIXTH(600),
		SEVENTH(700)
	}

	private fun ordinalFlow() = flow {
		for (ordinal in Ordinal.values()) {
			delay(ordinal.delay)
			emit(ordinal)
		}
	}

	@Test
	fun asyncBufferFlowTest() {
		val blockTime = measureTimeMillis {
			runBlocking {
				val timer = measureTimeMillis {
					ordinalFlow()
						.buffer()
						.collect { println("Collected: $it") }
				}
				println("Done in $timer")
			}
		}
		println("Block time in $blockTime ms")
	}

Testing the new

limitedParallelism

slicing I am not sure I fully understand how it works. Looking at the example below I am launching

Access #1

and a second later

Access #2

. Using the

accessDispatcher

with limited parallelism of 1 I would think the

Access #2

could only start after the

Access #1

was finished. So I expected the prints to go 1,2,3 but they go 1,3,2 instead. Could anyone point out why this is the actual behaviour? I know this can be solved with something like a Mutex, but I am trying to understand the behaviour of these new sliced dispatchers 😄

private val mainScope = MainScope()
  private val accessDispatcher = Dispatchers.IO.limitedParallelism(1)

  public fun foo() {
    mainScope.launch {
      mainScope.launch(accessDispatcher + CoroutineName("Access #1")) {
        println("Step 1")
        delay(5.seconds)
        println("Step 2")
      }
      delay(1.seconds)
      mainScope.launch(accessDispatcher + CoroutineName("Access #2")) {
        println("Step 3")
      }
    }
  }

If I can't use the SystemClock in my unit tests, then what's the point of the time value in

advanceTimeBy

? For example, I'm using a fake timesource so the actual advanced time doesn't matter.