Morning :smile: I currently have a sequential oper...

# announcements

[Edoardo Luppi]

Morning 😄
I currently have a sequential operation which uses a Sequence<T>

myProviders
  .asSequence()
  .sortedBy(configService::getProviderOrder)
  .flatMap { it.getCommitTypes(context.type) }
  .map { CommitTypePsiElement(project, it) }
  .mapIndexed(::CommitTypeLookupElement)
  .distinctBy(CommitTypeLookupElement::getLookupString)
  .forEach(rs::addElement)

Each provider call (it.getCommitTypes(context.type)) returns a Collection<T> and cannot be converted to Flows.
I would like to call every provider asynchronously and then flatMap the results, and collect .
How would you approach this in the code above?

This part

.map { CommitTypePsiElement(project, it) }
.mapIndexed(::CommitTypeLookupElement)
.distinctBy(CommitTypeLookupElement::getLookupString)
.forEach(rs::addElement)

can remain sequential / blocking.

[marstran]

Maybe something like this?

coroutineScope {
  myProviders
    .sortedBy(configService::getProviderOrder)
    .map{ async { it.getCommitTypes(context.type) } }
    .awaitAll()
    .asSequence()
    .flatten()
    .map { CommitTypePsiElement(project, it) }
    .mapIndexed(::CommitTypeLookupElement)
    .distinctBy(CommitTypeLookupElement::getLookupString)
    .forEach(rs::addElement)
}

[Edoardo Luppi]

@marstran mmmh, interesting, thanks!
Just to understand, what would happen if one call to getCommitTypes throws? The entire chain would fail, right?

[marstran]

Yes. It will throw on awaitAll . Those which hasn't completed yet will be cancelled.

[Edoardo Luppi]

@marstran okay! I will try to come up with a version which doesn't abort the other operations. This has been a good starting point 😉

[Edoardo Luppi]

@marstran Sorry if I bother you again!
The example you gave me works perfectly and I adapted it.
Now, after reading more on Coroutines and Flows I was trying to move to Flows (I'm used to RxJS).
I've come up with

runBlocking {
  myProviders
    .sortedBy(configService::getProviderOrder)
    .asFlow()
    .flatMapMerge { it.getCommitTypes(...).asFlow() }
    .onErrorCollect(emptyFlow())
    .flowOn(Dispatchers.Default)
    .toList()
    .map { CommitTypePsiElement(project, it) }
    .mapIndexed(::CommitTypeLookupElement)
    .distinctBy(CommitTypeLookupElement::getLookupString)
    .forEach(rs::addElement)
}

But it seems the calls are still blocking.

[Edoardo Luppi]

Oh btw, let me know if you prefer this to be on StackOverflow.

[Edoardo Luppi]

@marstran yes that's true. But doesn't even your version do it?

[Edoardo Luppi]

Or the await all considers the order of the DeferredResults?

[marstran]

Sorry, I deleted the comment. 😛
My original version preserves the order.

[marstran]

Yes, it does

[Edoardo Luppi]

@marstran oh okay! Yeah, I would need a concat to preserve the order, but then I lose the asynchronicity I think

[marstran]

Yes, that's what I was thinking as well.

[Edoardo Luppi]

At least the docs says concat wait for the coroutine to end before starting another one

[Edoardo Luppi]

But I'd need to come up with a better sorting strategy anyway, so let's not focus on the order 😄

[marstran]

Correct.
You may be able to use buffer to get around it though. Something like this:

myProviders
    .sortedBy(configService::getProviderOrder)
    .asFlow()
    .map { 
      flow { 
        emit(it.getCommitTypes()) 
      }.onErrorCollect(emptyFlow()) 
    }
    .buffer(10)
    .flattenConcat()
    .flatMapConcat { it.asFlow() }
    .flowOn(Dispatchers.Default)
    .toList()
    .map { CommitTypePsiElement(project, it) }
    .mapIndexed(::CommitTypeLookupElement)
    .distinctBy(CommitTypeLookupElement::getLookupString)
    .forEach(rs::addElement)

[marstran]

Not sure if it works though. It's getting kinda complex 😛

[marstran]

I think the async version was cleaner. I don't think it makes really sense to use a lazy Flow when you want all the elements to be computed in parallel anyway. We're kinda trying to force it to be eager.

[Edoardo Luppi]

@marstran Thanks! So to give you more context, I'm running this inside an IntelliJ Plugin. Currently as I said above I'm using a sequence to call those Providers.

To test coroutines "fast" I've done this

fun fillResultSetWithTypes(context: TypeCommitContext) {
  val rs = resultSet.withPrefixMatcher(context.type)
  safelyReleaseSemaphore(parameters.process)

  launch {
    val currentTimeMillis = System.currentTimeMillis()
    TYPE_EP.getExtensions(project)
      .sortedBy(configService::getProviderOrder)
      .asFlow()
      .map {
        flow {
          emit(it.getCommitTypes(context.type))
        }.onErrorCollect(emptyFlow())
      }
      .flattenConcat()
      .flatMapConcat { it.asFlow() }
      .flowOn(Dispatchers.Default)
      .toList()
      .map { CommitTypePsiElement(project, it) }
      .mapIndexed(::CommitTypeLookupElement)
      .distinctBy(CommitTypeLookupElement::getLookupString)
      .forEach(rs::addElement)
    val l = System.currentTimeMillis() - currentTimeMillis
    printlnError("Time: $l")
  }
}

With two fake Providers which call Thread.sleep(2000) the final time is always 4000+ ms.
Only the async versions did actually work

[Edoardo Luppi]

And yeah I understand the first version was cleaner. Just experimenting 😉

[Edoardo Luppi]

As a last resort, I will go back to the "old" (let me say that) Java concurrency.

[Edoardo Luppi]

@marstran oh btw, flattenConcat is what screws it up. With flattenMerge it works.

[marstran]

But you're missing buffer.

[Edoardo Luppi]

@marstran yeah I can't use it, I'm on 1.2.21 and it seems missing

[marstran]

Oh, ok

[Edoardo Luppi]

Although I don't understand why flatMapMerge doesn't work...

.flatMapMerge { it.getCommitTypes(...).asFlow() }

[Edoardo Luppi]

Or even what's the difference between

.map {
  flow {
    emit(it.getCommitTypes(context.type))
  }.onErrorCollect(emptyFlow())
}

And

.map {
  flowOf(it.getCommitTypes(context.type))
     .onErrorCollect(emptyFlow())
}

[marstran]

Using asFlow or flowOf will first eagerly execute it.getCommitTypes , and then create a flow from it.

[marstran]

flow { emit(x) } will create a flow immediately and only emit items when it.getCommitTypes is done.

[marstran]

So in the first case, flatMapMerge will actually perform all actions, when it only wants to "trigger" it (which it does by collecting the nested flow).

[marstran]

Do you see what I mean?

[Edoardo Luppi]

@marstran I'm reading and trying to understand 😄

Looking at flowOf I see

public fun <T> flowOf(vararg elements: T): Flow<T> = 
  unsafeFlow {
    for (element in elements) {
        emit(element)
    }
  }

Which is basically the same as

flow {
  emit(it.getCommitTypes(context.type))
}

[marstran]

With asFlow or flowOf , you get a Flow<Flow<T>> , but all the work has already been done.
With flow { emit(it) } , you get a Flow<Flow<T>> , but no work has been done yet.

[marstran]

No, it's not. Because it.getCommitTypes() has already been called when you call flowOf . The result is passed by value.

[Edoardo Luppi]

@marstran Oh heck... That's the basics. Sorry

[Edoardo Luppi]

Yeah I realized now hahaha

[marstran]

🙂

[Edoardo Luppi]

Let me try again to run that code!

[Edoardo Luppi]

@marstran yeah it works, as you said 🙂

[Edoardo Luppi]

I wanted to try the Flow approach mostly because of the fluid error handling with onError*

[marstran]

👍

[Edoardo Luppi]

@marstran Ok, I think I've got how everything works hahaha
But damn, is using a try - catch block the only way to handle an error using awaitAll ?

[Edoardo Luppi]

I miss the good CompletableFuture#exceptionally here!