diff options
author | Jakob Odersky <jakob@odersky.com> | 2016-04-18 02:34:57 -0700 |
---|---|---|
committer | Jakob Odersky <jakob@odersky.com> | 2016-04-18 02:38:54 -0700 |
commit | 3a1b65b9c4f71d58b9f0e62c0052030aa38fb00d (patch) | |
tree | d6fda5890153c17ffaebbb6cc5c7d061be00d70a /flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala | |
parent | 36721ebc573e1bb5f3e3ab74ed2feb0f6bf0b6e7 (diff) | |
download | akka-serial-3a1b65b9c4f71d58b9f0e62c0052030aa38fb00d.tar.gz akka-serial-3a1b65b9c4f71d58b9f0e62c0052030aa38fb00d.tar.bz2 akka-serial-3a1b65b9c4f71d58b9f0e62c0052030aa38fb00d.zip |
Implement stream interface for watchers
Diffstat (limited to 'flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala')
-rw-r--r-- | flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala | 38 |
1 files changed, 38 insertions, 0 deletions
diff --git a/flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala b/flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala new file mode 100644 index 0000000..e236f5d --- /dev/null +++ b/flow-stream/src/main/scala/com/github/jodersky/flow/stream/impl/WatcherStage.scala @@ -0,0 +1,38 @@ +package com.github.jodersky.flow +package stream +package impl + +import scala.concurrent.{Future, Promise} + +import akka.actor.ActorRef +import akka.stream.{Attributes, Outlet, SourceShape} +import akka.stream.stage.{GraphStageWithMaterializedValue, GraphStageLogic} + + +private[stream] class WatcherStage( + ioManager: ActorRef, + ports: Set[String] +) extends GraphStageWithMaterializedValue[SourceShape[String], Future[Serial.Watch]] { + + val out = Outlet[String]("Watcher.out") + + val shape = new SourceShape(out) + + override def createLogicAndMaterializedValue(attributes: Attributes): + (GraphStageLogic, Future[Serial.Watch]) = { + + val promise = Promise[Serial.Watch] + + val logic = new WatcherLogic( + shape, + ioManager, + ports, + promise + ) + + (logic, promise.future) + } + + override def toString = s"Watcher($ports)" + +} |