aboutsummaryrefslogtreecommitdiff
path: root/flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala
diff options
context:
space:
mode:
authorJakob Odersky <jakob@odersky.com>2016-05-16 15:28:59 -0700
committerJakob Odersky <jakob@odersky.com>2016-06-09 03:30:35 -0700
commit92c4b3d41e06ad4b89004212c85248e9e6cd61d7 (patch)
tree69470f7c4ed48edaebea91964d7d552e7eaacf0d /flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala
parentf6f26c2c9e3ec9bdd45fb384483b3450bef5984a (diff)
downloadakka-serial-92c4b3d41e06ad4b89004212c85248e9e6cd61d7.tar.gz
akka-serial-92c4b3d41e06ad4b89004212c85248e9e6cd61d7.tar.bz2
akka-serial-92c4b3d41e06ad4b89004212c85248e9e6cd61d7.zip
Move project to `ch.jodersky` and upgrade sbt-jni
Diffstat (limited to 'flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala')
-rw-r--r--flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala38
1 files changed, 38 insertions, 0 deletions
diff --git a/flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala b/flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala
new file mode 100644
index 0000000..82fad69
--- /dev/null
+++ b/flow-stream/src/main/scala/ch/jodersky/flow/stream/impl/WatcherStage.scala
@@ -0,0 +1,38 @@
+package ch.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)"
+
+}