1 files changed, 38 insertions, 0 deletions
diff --git a/stream/src/main/scala/akka/serial/stream/impl/WatcherStage.scala b/stream/src/main/scala/akka/serial/stream/impl/WatcherStage.scala
new file mode 100644
index 0000000..649249f
--- /dev/null
+++ b/stream/src/main/scala/akka/serial/stream/impl/WatcherStage.scala
@@ -0,0 +1,38 @@
+package akka.serial
+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)"
+
+}