1 files changed, 65 insertions, 0 deletions
diff --git a/stream/src/main/scala/akka/serial/stream/impl/WatcherLogic.scala b/stream/src/main/scala/akka/serial/stream/impl/WatcherLogic.scala
new file mode 100644
index 0000000..afab60e
--- /dev/null
+++ b/stream/src/main/scala/akka/serial/stream/impl/WatcherLogic.scala
@@ -0,0 +1,65 @@
+package akka.serial
+package stream
+package impl
+
+import scala.concurrent.Promise
+
+import akka.actor.{ActorRef, Terminated}
+import akka.stream.SourceShape
+import akka.stream.stage.GraphStageLogic
+import akka.serial.{Serial => CoreSerial}
+
+private[stream] class WatcherLogic(
+  shape: SourceShape[String],
+  ioManager: ActorRef,
+  ports: Set[String],
+  watchPromise: Promise[Serial.Watch])
+    extends GraphStageLogic(shape) {
+  import GraphStageLogic._
+
+  implicit private def self = stageActor.ref
+
+  override def preStart(): Unit = {
+    getStageActor(receive)
+    stageActor watch ioManager
+    for (dir <- WatcherLogic.getDirs(ports)) {
+      ioManager ! CoreSerial.Watch(dir, skipInitial = false)
+    }
+  }
+
+  setHandler(shape.out, IgnoreTerminateOutput)
+
+  private def receive(event: (ActorRef, Any)): Unit = {
+    val sender = event._1
+    val message = event._2
+
+    message match {
+
+      case Terminated(`ioManager`) =>
+        val ex = new StreamWatcherException("The serial IO manager has terminated. Stopping now.")
+        failStage(ex)
+        watchPromise.failure(ex)
+
+      case CoreSerial.CommandFailed(cmd, reason) =>
+        val ex = new StreamWatcherException(s"Serial command [$cmd] failed", reason)
+        failStage(ex)
+        watchPromise.failure(ex)
+
+      case CoreSerial.Connected(port) =>
+        if (ports contains port) {
+          if (isAvailable(shape.out)) {
+            push(shape.out, port)
+          }
+        }
+
+      case other =>
+        failStage(new StreamWatcherException(s"Stage actor received unkown message [$other]"))
+
+    }
+  }
+
+}
+
+private[stream] object WatcherLogic {
+  def getDirs(ports: Set[String]): Set[String] = ports.map(_.split("/").init.mkString("/"))
+}