SourceWithContext and FlowWithContext

[RayRoestenburg]

Based on earlier work by Johannes

• More docs
• Java version

I have added ApiMayChange @patriknw @jrudolph and processed @2m comments. Let me know if anything more is required for now.

[akka-ci]

Thank you for your pull request! After a quick sanity check one of the team will reply with 'OK TO TEST' to kick off our automated validation on Jenkins. This compiles the project, runs the tests, and checks for things like binary compatibility and source code formatting. When two team members have also manually reviewed and (perhaps after asking for some amendments) accepted your contribution, it should be good to be merged.

For more details about our contributing process, check out CONTRIBUTING.md - and feel free to ask!

[chbatey]

OK TO TEST

[akka-ci]

Test PASSed.

[akka-ci]

Test PASSed.

[akka-ci]

Test PASSed.

[akka-ci]

Test PASSed.

[akka-ci]

Test FAILed.

[akka-ci]

Test FAILed.

[jrudolph]

I cleaned up things and added scaladoc everywhere. From my side this is ready to go with this first set of supported operations.

[RayRoestenburg]

nice!

[RayRoestenburg]

@jrudolph nice simplifications and cleanups!

[akka-ci]

Test PASSed.

[patriknw]

LGTM, I have asked @RayRoestenburg to squash into 2 commits (Ray/Johannes) and then we can merge this.

[akka-ci]

Test PASSed.

[patriknw]

Great work @RayRoestenburg and @jrudolph

[He-Pin]

nice too

[jroper]

Sorry for the late review, I only just learnt about this.

I suspect a common pattern might be where users don't care about what the context is, they just want a flow that carries whatever context the container (eg, pipelines, or Akka projections) wants to provide. Then, the same flow can be used in multiple situations with different contexts, eg carrying a Kafka context in production, or carrying some dummy context in tests.

An example would be in the Akka projections library, when you define a projection, the context is irrelevant. With this API, I could imagine Akka projections might provide something that looks like this:

trait Projection[In, Out] {
  def projection[Ctx]: FlowWithContext[In, Ctx, Out, Ctx, _]
}

And then the user would implement it like this:

class MyProjection extends Projection[MyIn, MyOut] {
  override def projection[Ctx] = FlowWithContext[MyIn, Ctx].map(myInToMyOut)
}

The question for me then is, can we possibly drop the context type variable somehow? It's not so bad for Scala, but for Java the signatures have the potential to be quite overbearing. I realise that the reason the context type variable is needed is to provide via, but if we made a decision that the context is always opaque within a FlowWithContext, then via could potentially be provided via some mechanism like this:

trait ViaWithContext[In, Out, Mat] {
  def apply[Ctx]: Flow[(In, Ctx), (Out, Ctx), Mat]
}

def via[Out, Mat](flow: ViaWithContext[In, Out, Mat]): FlowWithContext[In, Out, Mat]

And maybe that could be implemented with a lambda because it's a SAM? The same might work in Java? I'm not sure, the ViaWithContext trait only exists essentially as a type constructor, which might be a little confusing to users, but I think the source code should look simple enough.

[jrudolph]

@jroper

I agree that a FlowWithContext without a fixed context would be a good thing. I had a try at getting that to work but it was non-trivial to get right so we settled for the straight-forward API for now (even if that means extra type parameters).

In any case, we'd like to keep the API surface minimal. If we want a generic FlowWithContext that should be the only API and we should be able to implement it without having to copy all implementations from FlowWithContextOps.

If we can achieve that, I'm all for doing it, it's not to late to iterate on the API (and @raboof also had ideas for improvements he wants to try out at some point).