akka/akka-core

Amend documentation about adding custom operations on FlowOps

Open

#21,856 opened on Nov 17, 2016

View on GitHub
 (3 comments) (0 reactions) (0 assignees)Scala (3,547 forks)batch import
1 - triagedhelp wantedt:docst:stream

Repository metrics

Stars
 (13,277 stars)
PR merge metrics
 (Avg merge 8d 19h) (10 merged PRs in 30d)

Description

With Scala 2.12 the crazy type-level-hack-workaround as discovered by @rkuhn back in the days is not needed anymore. We pointed it out here: http://doc.akka.io/docs/akka/rp-current/scala/stream/stream-customize.html#Extending_Flow_Combinators_with_Custom_Operators

With 2.12 we can show off that:


scala> import akka.stream._
import akka.stream._

scala> import akka.stream.scaladsl._
import akka.stream.scaladsl._

scala> import akka.stream.stage._
import akka.stream.stage._

scala> import scala.language.higherKinds
import scala.language.higherKinds

scala> final case class Klangified[T](val the: T)
defined class Klangified

scala> final implicit class KlangifyOps[O, M, FO[o,m] <: FlowOps[o,m]](val fo: FO[O, M]) {
     |   type Repr[T] = FO[O, M]#Repr[T]
     |   def klangify(times: Int): Repr[Klangified[O]] = fo.mapConcat(x => List.fill(times)(Klangified(x)))
     |   def passthru: Repr[O] = fo.map(identity)
     | }
defined class KlangifyOps

scala> Source(1 to 10).passthru
res11: akka.stream.scaladsl.Source[Int,akka.NotUsed] =
Source(SourceShape(Map.out), CompositeModule [23661f5d]
  Name: unnamed
  Modules:
    (iterableSource) CompositeModule [546cd17e]
      Name: iterableSource
      Modules:
        (singleSource) GraphStage(SingleSource(Range 1 to 10)) [5fb58725]
        (unnamed) [6286e224] copy of GraphStage(StatefulMapConcat) [710995de]
      Downstreams:
        single.out -> StatefulMapConcat.in
      Upstreams:
        StatefulMapConcat.in -> single.out
      MatValue: Atomic(singleSource[5fb58725])
    (unnamed) [435c4da0] copy of GraphStage(Map(KlangifyOps$$Lambda$2787/1867139207@65e135b3)) [30a36641]
  Downstreams:
    single.out -> StatefulMapConcat.in
    StatefulMapConcat.out -> Map.in
  Upstreams:
    StatefulMapConcat.in -> single.out
...
scala> Flow[Int].passthru
res12: akka.stream.scaladsl.Flow[Int,Int,akka.NotUsed] =
Flow(FlowShape(Map.in,Map.out), CompositeModule [13e890d0]
  Name: unnamed
  Modules:
    (map) GraphStage(Map(KlangifyOps$$Lambda$2787/1867139207@65e135b3)) [45280b87]
  Downstreams:
  Upstreams:
  MatValue: Ignore)

scala> res12.klangify(2)
res13: akka.stream.scaladsl.Flow[Int,Klangified[Int],akka.NotUsed] =
Flow(FlowShape(Map.in,StatefulMapConcat.out), CompositeModule [637ad325]
  Name: unnamed
  Modules:
    (map) GraphStage(Map(KlangifyOps$$Lambda$2787/1867139207@65e135b3)) [45280b87]
    (unnamed) [5259b832] copy of GraphStage(StatefulMapConcat) [2146991d]
  Downstreams:
    Map.out -> StatefulMapConcat.in
  Upstreams:
    StatefulMapConcat.in -> Map.out
  MatValue: Ignore)

scala> res11.klangify(2)
res14: akka.stream.scaladsl.Source[Klangified[Int],akka.NotUsed] =
Source(SourceShape(StatefulMapConcat.out), CompositeModule [464d5e11]
...

via @viktorklang

Contributor guide