akka/akka-core
View on GitHubAmend documentation about adding custom operations on FlowOps
Open
#21,856 opened on Nov 17, 2016
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