str feat: RestartSourceWithContext

[leviramsey]

SourceWithContext cannot be wrapped into a RestartSource.

val sourceWithContext = SourceWithContext.fromTuples(Source(Seq("A", "B", "C").zipWithIndex))

RestartSource.onFailuresWithBackoff(RestartSettings(1.milli, 10.millis, 0.2)) { () => sourceWithContext }

       error: type mismatch;
        found   : akka.stream.scaladsl.SourceWithContext[String,Int,akka.NotUsed]
        required: akka.stream.scaladsl.Source[?, _]

I would assume that similar limitations apply to SinkWithContext/FlowWithContext, but it's not clear how well their respective limitations with respect to dropping on restart mesh with the intended use of the ...WithContext variations for carrying something like a Kafka or Projection offset, which tend to be where an at-least-once guarantee is desired.

[leviramsey]

Weird, I can't even get Test / compile to work for me with Scala 3.3 due to an akka-(actor|cluster)-typed weirdness (so not brought in by this change)

akka > +~ 3.3 Test/compile
[error] /home2/levi/code/akka/akka-actor-typed/src/main/scala-3/akka/actor/typed/internal/receptionist/Platform.scala:16: error: ; expected but match found
[error]   type Service[K <: Aux[_]] = K match {
[error]                                 ^: /home2/levi/code/akka/akka-actor-typed/src/main/scala-3/akka/actor/typed/internal/receptionist/Platform.scala
[error] /home2/levi/code/akka/akka-cluster-typed/src/main/scala-3/akka/cluster/typed/internal/receptionist/ClusterReceptionistProtocol.scala:18: error: ; expected but match found
[error]   type SubscriptionsKV[K <: Aux[_]] = K match {
[error]                                         ^: /home2/levi/code/akka/akka-cluster-typed/src/main/scala-3/akka/cluster/typed/internal/receptionist/ClusterReceptionistProtocol.scala
[error] (akka-actor-typed / Compile / scalafmt) /home2/levi/code/akka/akka-actor-typed/src/main/scala-3/akka/actor/typed/internal/receptionist/Platform.scala:16: error: ; expected but match found
[error]   type Service[K <: Aux[_]] = K match {
[error]                                 ^: /home2/levi/code/akka/akka-actor-typed/src/main/scala-3/akka/actor/typed/internal/receptionist/Platform.scala
[error] (akka-cluster-typed / Compile / scalafmt) /home2/levi/code/akka/akka-cluster-typed/src/main/scala-3/akka/cluster/typed/internal/receptionist/ClusterReceptionistProtocol.scala:18: error: ; expected but match found
[error]   type SubscriptionsKV[K <: Aux[_]] = K match {
[error]                                         ^: /home2/levi/code/akka/akka-cluster-typed/src/main/scala-3/akka/cluster/typed/internal/receptionist/ClusterReceptionistProtocol.scala

EDIT: scalafmt (see #32186 & #32187)

[leviramsey]

Close/reopening for CI

[johanandren]

Looks good, needs Java APIs

[johanandren]

One more thing then I think this is ready to go.

[johanandren]

The akka.japi.function interfaces are parallel with the JDK stdlib ones but add the capability to throw checked, for example for usage in actor message handling. That's not important here, so let's go with java.util.function.Supplier instead.

[leviramsey]

Only used Creator to mirror the other Restart* factories... suppose there should be an issue to move those to Supplier (which would be binary incompat, but source compat) in 2.10?

[johanandren]

Ah, yes, let's keep it aligned then 👍

[johanandren]

LGTM