add error management

4 files changed, 33 insertions, 26 deletions

diff --git a/samples/rwc/src/main/scala/com/github/jodersky/flow/example/Main.scala b/samples/rwc/src/main/scala/com/github/jodersky/flow/example/Main.scala
index 985513b..518efc2 100644
--- a/samples/rwc/src/main/scala/com/github/jodersky/flow/example/Main.scala
+++ b/samples/rwc/src/main/scala/com/github/jodersky/flow/example/Main.scala
@@ -1,20 +1,15 @@
 package com.github.jodersky.flow
 package example
 
-import scala.concurrent.ExecutionContext.Implicits.global
-import scala.util.Success
-import scala.util.Try
-import com.github.jodersky.flow.Serial
-import com.github.jodersky.flow.Serial._
 import akka.actor.ActorSystem
 import akka.actor.Props
-import akka.io.IO
+import akka.actor.actorRef2Scala
 import akka.util.ByteString
-import com.github.jodersky.flow.internal.InternalSerial
 
 object Main {
 
   def main(args: Array[String]): Unit = {
+    
     /*val isInt = Try(args(1).toInt) match { case Success(_) => true; case _ => false }
     if (!(args.length == 2 && isInt)) {
       println("invalid parameters")
@@ -25,18 +20,18 @@ object Main {
     val port = "/dev/ttyACM0"
     val baud = 115200
 
-   // InternalSerial.debug(true)
-    
+    // InternalSerial.debug(true)
+
     implicit val system = ActorSystem("flow")
     val serial = system.actorOf(Props(classOf[SerialHandler], port, baud), name = "serial-handler")
-    
+
     readLine()
     serial ! ByteString("hello back".getBytes())
-    
+
     readLine()
     serial ! "close"
     readLine()
-    
+
     system.shutdown()
   }
 }
\ No newline at end of file
diff --git a/samples/rwc/src/main/scala/com/github/jodersky/flow/example/SerialHandler.scala b/samples/rwc/src/main/scala/com/github/jodersky/flow/example/SerialHandler.scala
index ddcde8d..e616763 100644
--- a/samples/rwc/src/main/scala/com/github/jodersky/flow/example/SerialHandler.scala
+++ b/samples/rwc/src/main/scala/com/github/jodersky/flow/example/SerialHandler.scala
@@ -24,23 +24,23 @@ class SerialHandler(port: String, baud: Int) extends Actor with ActorLogging {
 
     case Opened(port) => {
       log.info(s"Port ${port} is now open.")
-      context watch sender
       context become opened(sender)
     }
   }
 
   def opened(operator: ActorRef): Receive = {
-    case Terminated(`operator`) => {
-      log.info("operator down, handler exiting")
-      context.stop(self)
-    }
+   
     case Received(data) => {
       log.info("Received data: " + formatData(data))
       log.info("As string: " + new String(data.toArray, "UTF-8"))
     }
     case Wrote(data) => log.info("Got ACK for writing data: " + formatData(data))
-    case Closed => {
-      log.info("Operator closed, exiting handler.")
+    case Closed(None) => {
+      log.info("Operator closed normally, exiting handler.")
+      context stop self
+    }
+    case Closed(Some(ex)) => {
+      log.info("Operator crashed, exiting handler.")
       context stop self
     }
     case "close" => {
diff --git a/src/main/scala/com/github/jodersky/flow/Serial.scala b/src/main/scala/com/github/jodersky/flow/Serial.scala
index 6f18e19..6e4aefb 100644
--- a/src/main/scala/com/github/jodersky/flow/Serial.scala
+++ b/src/main/scala/com/github/jodersky/flow/Serial.scala
@@ -20,6 +20,6 @@ object Serial extends ExtensionKey[SerialExt] {
   case class Wrote(data: ByteString) extends Event
   
   case object Close extends Command
-  case object Closed extends Event
+  case class Closed(error: Option[Exception]) extends Event
   
 }
diff --git a/src/main/scala/com/github/jodersky/flow/SerialOperator.scala b/src/main/scala/com/github/jodersky/flow/SerialOperator.scala
index 8621c55..604dc74 100644
--- a/src/main/scala/com/github/jodersky/flow/SerialOperator.scala
+++ b/src/main/scala/com/github/jodersky/flow/SerialOperator.scala
@@ -1,5 +1,7 @@
 package com.github.jodersky.flow
 
+import java.io.IOException
+
 import com.github.jodersky.flow.internal.InternalSerial
 
 import Serial.Close
@@ -18,7 +20,7 @@ import akka.util.ByteString
 class SerialOperator(handler: ActorRef, serial: InternalSerial) extends Actor with ActorLogging {
   import context._
 
-  case class ReadException(ex: Throwable)
+  case class ReadException(ex: Exception)
 
   object Reader extends Thread {
 
@@ -68,19 +70,29 @@ class SerialOperator(handler: ActorRef, serial: InternalSerial) extends Actor wi
   def receive: Receive = {
 
     case Write(data, ack) => {
-      serial.write(data.toArray) // no future needed as write is non-blocking
-      if (ack) sender ! Wrote(data)
+      try {
+        val sent = serial.write(data.toArray)
+        if (ack) sender ! Wrote(ByteString(sent))
+      } catch {
+        case ex: IOException => {
+          handler ! Closed(Some(ex))
+          context stop self
+        }
+      }
     }
 
     case Close => {
-      sender ! Closed
-      context.stop(self)
+      handler ! Closed(None)
+      context stop self
     }
 
     case Terminated(`handler`) => context.stop(self)
 
     //go down with reader thread
-    case ReadException(ex) => throw ex
+    case ReadException(ex) => {
+      handler ! Closed(Some(ex))
+      context stop self
+    }
 
   }