raboof/pekko
1
0
Fork 0
Code Issues Pull requests Projects Releases 6 Packages Wiki Activity Actions 10

Refactor anon GraphStageWithMaterializedValue to object

Browse source
This commit is contained in:
Bernard Leach 2016-04-06 23:28:48 +10:00
parent 7357571c82
commit 174f812e99
1 changed files with 6 additions and 5 deletions
Download patch file Download diff file Expand all files Collapse all files

11
akka-http-core/src/main/scala/akka/http/impl/util/StreamUtils.scala
View file

@ -226,9 +226,9 @@ private[http] object StreamUtils {
* is finished, only that the part that contains this flow has finished work. * is finished, only that the part that contains this flow has finished work.
*/ */
def identityFinishReporter[T]: Flow[T, T, Future[Unit]] = { def identityFinishReporter[T]: Flow[T, T, Future[Unit]] = {
Flow[T].viaMat(new GraphStageWithMaterializedValue[FlowShape[T, T], Future[Unit]] { object IdentityFinishReporter extends GraphStageWithMaterializedValue[FlowShape[T, T], Future[Unit]] {
val shape = FlowShape(Inlet[T]("identityFinishReporter.in"), Outlet[T]("identityFinishReporter.out")) val shape = FlowShape(Inlet[T]("identityFinishReporter.in"), Outlet[T]("identityFinishReporter.out"))
override def toString: String = "UniqueKillSwitchFlow" override def toString: String = "IdentityFinishReporter"
def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Unit]) = { def createLogicAndMaterializedValue(inheritedAttributes: Attributes): (GraphStageLogic, Future[Unit]) = {
val promise = Promise[Unit]() val promise = Promise[Unit]()
@ -247,13 +247,14 @@ private[http] object StreamUtils {
promise.trySuccess(()) promise.trySuccess(())
} }
setHandler(shape.in, this) setHandlers(shape.in, shape.out, this)
setHandler(shape.out, this)
} }
(stage, promise.future) (stage, promise.future)
} }
})(Keep.right) }
Flow[T].viaMat(IdentityFinishReporter)(Keep.right)
} }
/** /**