Can I (and should I) emit to a Flow from multiple threads? For example, mapping a list of objects and making an HTTP request using the IO Dispatcher, and emitting the result one by one.

ids.map { id ->
	async(<http://Dispatchers.IO|Dispatchers.IO>) {
		api.getMember(id) // I want to emit this
	}
}

Or is there a better-suited construct for this, like a channel?

I'm not sure where the Flow factors into this? I do think that FlowCollector.emit isn't thread safe, but you can probably batch up the Deferred results and then collect them in a single thread/coroutine for emission, e.g.

flow.map { async { doSomethingToMakeList(it) } }
  .buffer(8)
  .map { it.await() }
  .flatMap { it.asFlow() }

getMember

doesn't return a Flow, it's a suspending function that returns a data object. What's your

flow

object in your first response?

it's just a Flow<T>, of whatever you want it to be. If you're mapping a list, I would just call

asFlow()

on it, then you can do the map-async-buffer-map like above

I see. I'm still a bit confused on what is "better," because there's also the

produce

strategy:

private suspend fun loadAll0(ids: List<String>) = coroutineScope {
		produce {
			ids.forEach { id ->
				async(<http://Dispatchers.IO|Dispatchers.IO>) {
					send(api.getMember(id))
				}
			}
		}
	}

yes, you can do that too.

asFlow()

just generates a "cold" stream instead, which won't start until a terminal operation like

collect

is called. If you want a "hot" stream, you can use

produce

+

consumeAsFlow()

iirc

Yeah, the other main advantage of Flow is the powerful extension methods and such that it offers. It's very tempting to try and use it wherever possible

the benefit of the map-async-buffer-map, is that it prevents too many tasks from running at once

Ah, that is a good point

right now you're launching a task for every id at once, which if you have >64/128 (don't recall exactly), you may fill the IO dispatcher threads. the

buffer

prevents that by only allowing the flow to have that many + 1 tasks running

emit should throw an exception if you change contexts.

channelFlow

might be what you're looking for (being able to emit from multiple jobs across different threads)