The dispatcher is used only for reading and closing. That seems sequential to me.

So this is interesting. Previously unsafeRunSync was necessary to achieve this behavior. Now the shared sequential Dispatcher guarantees the required ordering out-of-the box.

We need two Dispatchers for StreamSubscription, but they can both be sequential.

The first is used in unsafeStart() and actually runs the subscription. So it's blocked on that.

The second is used for handling requests and canceling.

Since we can now have local Dispatchers with special sequential semantics, it seems like we should deprecate the pattern of asking for (presumably more global) dispatchers in favor of building our own dispatchers.

Furthermore, since we now rely on the sequential Dispatcher to achieve the correct semantics, exposing it as an argument is not good design.

I'm sorry to come in to this so late in the game after this has merged, but we use this in a Play application to stream CSV files from an fs2 Stream; the Play API requires a class that's hard to construct except by using a StreamUnicastPublisher but it doesn't give us the opportunity to call close. In previous discussions on discord, I was advised to create a Dispatcher for the application and pass it in; we can guarantee the provided Dispatcher gets closed at the end.

I see (superficially) the point of your comment here and the rationale behind deprecating the API that takes a provided Dispatcher, but I don't see how we can migrate to the non-deprecated API without leaking Dispatcher instances. I will try to find you on discord sometime to talk about it a little more.

@mtomko no problem, thanks for the comment.

but I don't see how we can migrate to the non-deprecated API without leaking Dispatcher instances.

I need to think about it more, but so long as you are careful to cancel the subscription, I don't think there will be any leak here. The Resource surrounding a Dispatcher is intended to make sure that any tasks it started are canceled when the resource closes. If a Dispatcher has no running tasks and it goes out of scope, I am fairly certain that the GC will completely clean it up. Hence, no leak.

Also @BalmungSan is currently working on some usability improvements that might help your usecase :)

I'm trying to trace through the code, but it means digging deeply into akka streams and Play. The code we've written just creates the source and passes it to Play's Ok to construct a result:

  def streamToClientAsCsvFile(
      enumerator: Stream[IO, Row],
      dispatcher: Dispatcher[IO],
      filename: String
  ): Result = {
    val publisher = StreamUnicastPublisher(enumerator.through(rowToCsv), dispatcher)
    val source = Source.fromPublisher(publisher)

    Ok.chunked(source)
      .withHeaders(CONTENT_DISPOSITION -> s"attachment; filename = $filename")
      .as("text/csv")
  }

I would like to hope that Play is smart enough to recognize when the connection is closed and close the Source.

Ps, thanks for your previous response, it's really appreciated!

@mtomko #3107 is the PR @armanbilge mentioned. However, I am not exactly sure how it helps in your use case.

However, I am not exactly sure how it helps in your use case.

@BalmungSan @mtomko the new stream.subscribe(...) syntax that Luis is adding in #3107 will enable you to bring back the global Dispatcher.

val source = Source.fromPublisher { subscriber =>
  dispatcher.unsafeRunAndForget(enumerator.subscribe(subscriber))
}

This machinery was no longer necessary since Dispatcher is no longer accepted as an argument.

read/write are sequential and error is one-off, but they need to be independent, so they need their own dispatchers.

Super weird, can reproduce just by updating to 3.2.0 on main. No idea ...

I minimized it in armanbilge/sandbox@d9453ac. IDK what change in dotty caused it, but seems safe: seems like the private constructor was generating public bytecode prior to 3.2.0, but is now properly private. This is flagging up MiMa.

It doesn't affect anything public.