From 892bd6053aff21cda2fbabddb824c9710d7baafa Mon Sep 17 00:00:00 2001 From: Jakob Odersky Date: Tue, 18 Jun 2013 23:55:38 +0200 Subject: repair broken example --- .../scala/com/github/jodersky/flow/Framing.scala | 104 +++++++++++++++++++++ .../scala/com/github/jodersky/flow/Framing.scalac | 104 --------------------- .../scala/com/github/jodersky/flow/Serial.scala | 34 +++++++ .../scala/com/github/jodersky/flow/Serial.scalac | 34 ------- .../com/github/jodersky/flow/SerialManager.scala | 35 +++++++ .../com/github/jodersky/flow/SerialManager.scalac | 35 ------- .../com/github/jodersky/flow/SerialOperator.scala | 54 +++++++++++ .../com/github/jodersky/flow/SerialOperator.scalac | 44 --------- 8 files changed, 227 insertions(+), 217 deletions(-) create mode 100644 src/main/scala/com/github/jodersky/flow/Framing.scala delete mode 100644 src/main/scala/com/github/jodersky/flow/Framing.scalac create mode 100644 src/main/scala/com/github/jodersky/flow/Serial.scala delete mode 100644 src/main/scala/com/github/jodersky/flow/Serial.scalac create mode 100644 src/main/scala/com/github/jodersky/flow/SerialManager.scala delete mode 100644 src/main/scala/com/github/jodersky/flow/SerialManager.scalac create mode 100644 src/main/scala/com/github/jodersky/flow/SerialOperator.scala delete mode 100644 src/main/scala/com/github/jodersky/flow/SerialOperator.scalac (limited to 'src/main/scala') diff --git a/src/main/scala/com/github/jodersky/flow/Framing.scala b/src/main/scala/com/github/jodersky/flow/Framing.scala new file mode 100644 index 0000000..f8173a7 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/Framing.scala @@ -0,0 +1,104 @@ +package com.github.jodersky.flow + +import akka.io.PipelineContext +import akka.io.SymmetricPipePair +import akka.io.SymmetricPipelineStage +import akka.util.ByteString +import java.nio.ByteOrder +import scala.annotation.tailrec +import java.nio.ByteBuffer + +class DelimitedFrame( + StartByte: Byte, + StopByte: Byte, + EscapeByte: Byte) + //byteOrder: ByteOrder = ByteOrder.BIG_ENDIAN) + extends SymmetricPipelineStage[PipelineContext, ByteString, ByteString] { + + // range checks omitted ... + + override def apply(ctx: PipelineContext) = + new SymmetricPipePair[ByteString, ByteString] { + var buffer = ByteString.empty + //implicit val byteOrder = DelimitedFrame.this.byteOrder + + sealed trait State + case object Waiting extends State + case object Accepting extends State + case object Escaping extends State + + def extractFrame(bs: ByteString, accepted: ByteString, state: State): (ByteString, Option[ByteString]) = { //(remaining, frame)) + if (bs.isEmpty && state == Waiting) (ByteString.empty, None) + else if (bs.isEmpty) (accepted, None) + else { + val in = bs.head + + state match { + case Waiting if (in == StartByte) => extractFrame(bs.tail, accepted, Accepting) + case Escaping => extractFrame(bs.tail, accepted ++ ByteString(in), Accepting) + case Accepting => in match { + case EscapeByte => extractFrame(bs.tail, accepted, Escaping) + case StartByte => extractFrame(bs.tail, ByteString.empty, Accepting) + case StopByte => (bs.tail, Some(accepted)) + case other => extractFrame(bs.tail, accepted ++ ByteString(other), Accepting) + } + case _ => extractFrame(bs.tail, accepted, state) + } + } + } + + def extractFrames(bs: ByteString, accepted: List[ByteString]): (ByteString, List[ByteString]) = { + val (remainder, frame) = extractFrame(bs, ByteString.empty, Waiting) + + frame match { + case None => (remainder, accepted) + case Some(data) => extractFrames(remainder, data :: accepted) + } + } + + /* + * This is how commands (writes) are transformed: calculate length + * including header, write that to a ByteStringBuilder and append the + * payload data. The result is a single command (i.e. `Right(...)`). + */ + override def commandPipeline = { bs: ByteString => + val bb = ByteString.newBuilder + + def escape(b: Byte) = { + bb += EscapeByte + bb += b + } + + bb += StartByte + for (b <- bs) { + b match { + case StartByte => escape(b) + case StopByte => escape(b) + case EscapeByte => escape(b) + case _ => bb += b + } + } + bb += StopByte + + ctx.singleCommand(bb.result) + } + + /* + * This is how events (reads) are transformed: append the received + * ByteString to the buffer (if any) and extract the frames from the + * result. In the end store the new buffer contents and return the + * list of events (i.e. `Left(...)`). + */ + override def eventPipeline = { bs: ByteString => + val data = buffer ++ bs + val (remainder, frames) = extractFrames(data, Nil) + buffer = remainder + + frames match { + case Nil => Nil + case one :: Nil ⇒ ctx.singleEvent(one) + case many ⇒ many reverseMap (Left(_)) + } + } + } +} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/Framing.scalac b/src/main/scala/com/github/jodersky/flow/Framing.scalac deleted file mode 100644 index f8173a7..0000000 --- a/src/main/scala/com/github/jodersky/flow/Framing.scalac +++ /dev/null @@ -1,104 +0,0 @@ -package com.github.jodersky.flow - -import akka.io.PipelineContext -import akka.io.SymmetricPipePair -import akka.io.SymmetricPipelineStage -import akka.util.ByteString -import java.nio.ByteOrder -import scala.annotation.tailrec -import java.nio.ByteBuffer - -class DelimitedFrame( - StartByte: Byte, - StopByte: Byte, - EscapeByte: Byte) - //byteOrder: ByteOrder = ByteOrder.BIG_ENDIAN) - extends SymmetricPipelineStage[PipelineContext, ByteString, ByteString] { - - // range checks omitted ... - - override def apply(ctx: PipelineContext) = - new SymmetricPipePair[ByteString, ByteString] { - var buffer = ByteString.empty - //implicit val byteOrder = DelimitedFrame.this.byteOrder - - sealed trait State - case object Waiting extends State - case object Accepting extends State - case object Escaping extends State - - def extractFrame(bs: ByteString, accepted: ByteString, state: State): (ByteString, Option[ByteString]) = { //(remaining, frame)) - if (bs.isEmpty && state == Waiting) (ByteString.empty, None) - else if (bs.isEmpty) (accepted, None) - else { - val in = bs.head - - state match { - case Waiting if (in == StartByte) => extractFrame(bs.tail, accepted, Accepting) - case Escaping => extractFrame(bs.tail, accepted ++ ByteString(in), Accepting) - case Accepting => in match { - case EscapeByte => extractFrame(bs.tail, accepted, Escaping) - case StartByte => extractFrame(bs.tail, ByteString.empty, Accepting) - case StopByte => (bs.tail, Some(accepted)) - case other => extractFrame(bs.tail, accepted ++ ByteString(other), Accepting) - } - case _ => extractFrame(bs.tail, accepted, state) - } - } - } - - def extractFrames(bs: ByteString, accepted: List[ByteString]): (ByteString, List[ByteString]) = { - val (remainder, frame) = extractFrame(bs, ByteString.empty, Waiting) - - frame match { - case None => (remainder, accepted) - case Some(data) => extractFrames(remainder, data :: accepted) - } - } - - /* - * This is how commands (writes) are transformed: calculate length - * including header, write that to a ByteStringBuilder and append the - * payload data. The result is a single command (i.e. `Right(...)`). - */ - override def commandPipeline = { bs: ByteString => - val bb = ByteString.newBuilder - - def escape(b: Byte) = { - bb += EscapeByte - bb += b - } - - bb += StartByte - for (b <- bs) { - b match { - case StartByte => escape(b) - case StopByte => escape(b) - case EscapeByte => escape(b) - case _ => bb += b - } - } - bb += StopByte - - ctx.singleCommand(bb.result) - } - - /* - * This is how events (reads) are transformed: append the received - * ByteString to the buffer (if any) and extract the frames from the - * result. In the end store the new buffer contents and return the - * list of events (i.e. `Left(...)`). - */ - override def eventPipeline = { bs: ByteString => - val data = buffer ++ bs - val (remainder, frames) = extractFrames(data, Nil) - buffer = remainder - - frames match { - case Nil => Nil - case one :: Nil ⇒ ctx.singleEvent(one) - case many ⇒ many reverseMap (Left(_)) - } - } - } -} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/Serial.scala b/src/main/scala/com/github/jodersky/flow/Serial.scala new file mode 100644 index 0000000..1eacd6e --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/Serial.scala @@ -0,0 +1,34 @@ +package com.github.jodersky.flow + +import akka.io._ +import akka.actor.ExtensionKey +import akka.actor.ExtendedActorSystem +import akka.actor.Props +import low.{ Serial => LowSerial } +import akka.actor.ActorRef +import akka.util.ByteString + +object Serial extends ExtensionKey[SerialExt] { + + trait Command + trait Event + + case class Open(handler: ActorRef, port: String, baud: Int) extends Command + case class Opened(operator: ActorRef) extends Event + + case class Received(data: ByteString) extends Event + + case class Write(data: ByteString) extends Command + case class Wrote(data: ByteString) extends Event + + case object Close extends Command + case object Closed extends Event + + case class CommandFailed(command: Command, reason: Throwable) extends Event + + +} + +class SerialExt(system: ExtendedActorSystem) extends IO.Extension { + def manager = system.actorOf(Props[SerialManager], name = "IO-SERIAL") +} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/Serial.scalac b/src/main/scala/com/github/jodersky/flow/Serial.scalac deleted file mode 100644 index 7182425..0000000 --- a/src/main/scala/com/github/jodersky/flow/Serial.scalac +++ /dev/null @@ -1,34 +0,0 @@ -package com.github.jodersky.flow - -import akka.io._ -import akka.actor.ExtensionKey -import akka.actor.ExtendedActorSystem -import akka.actor.Props -import low.{ Serial => LowSerial } -import akka.actor.ActorRef -import akka.util.ByteString - -object Serial extends ExtensionKey[SerialExt] { - - trait Command - trait Event - - case class Open(handler: ActorRef, port: String, baud: Int) extends Command - case class Opened(operator: ActorRef) extends Event - - case class Received(data: ByteString) extends Event - - case class Write(data: ByteString) extends Command - case class Wrote(data: ByteString) extends Event - - case object Close extends Command - - - case class CommandFailed(command: Command, reason: Throwable) extends Event - - -} - -class SerialExt(system: ExtendedActorSystem) extends IO.Extension { - def manager = system.actorOf(Props[SerialManager], name = "IO-SERIAL") -} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/SerialManager.scala b/src/main/scala/com/github/jodersky/flow/SerialManager.scala new file mode 100644 index 0000000..4af2f9a --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/SerialManager.scala @@ -0,0 +1,35 @@ +package com.github.jodersky.flow + +import akka.actor.Actor +import Serial._ +import low.{ Serial => LowSerial } +import scala.util.Success +import scala.util.Failure +import akka.actor.Props +import scala.concurrent._ + +class SerialManager extends Actor { + import SerialManager._ + import context._ + + def receive = { + case command @ Open(handler, port, baud) => + future{LowSerial.open(port, baud)}.onComplete(_ match { + case Success(serial) => { + val operator = context.actorOf(Props(classOf[SerialOperator], serial, handler), name = escapePortString(port)) + handler ! Opened(operator) + } + case Failure(t) => handler ! CommandFailed(command, t) + }) + } + +} + +object SerialManager { + + private def escapePortString(port: String) = port collect { + case '/' => '-' + case c => c + } + +} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/SerialManager.scalac b/src/main/scala/com/github/jodersky/flow/SerialManager.scalac deleted file mode 100644 index ca3fc6b..0000000 --- a/src/main/scala/com/github/jodersky/flow/SerialManager.scalac +++ /dev/null @@ -1,35 +0,0 @@ -package com.github.jodersky.flow - -import akka.actor.Actor -import Serial._ -import low.{ Serial => LowSerial } -import scala.util.Success -import scala.util.Failure -import akka.actor.Props -import scala.concurrent._ - -class SerialManager extends Actor { - import SerialManager._ - import context._ - - def receive = { - case command @ Open(handler, port, baud) => - future{LowSerial.open(port, baud)}.onComplete(_ match { - case Success(serial) => { - val operator = context.actorOf(Props(classOf[SerialOperator], serial, handler), name = escapePortString(port)) - handler ! Opened(operator) - } - case Failure(t) => sender ! CommandFailed(command, t) - }) - } - -} - -object SerialManager { - - private def escapePortString(port: String) = port collect { - case '/' => '-' - case c => c - } - -} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/SerialOperator.scala b/src/main/scala/com/github/jodersky/flow/SerialOperator.scala new file mode 100644 index 0000000..6177188 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/SerialOperator.scala @@ -0,0 +1,54 @@ +package com.github.jodersky.flow + +import scala.util.Failure +import scala.util.Success +import Serial._ +import akka.actor.Actor +import akka.actor.ActorLogging +import akka.actor.ActorRef +import akka.util.ByteString +import low.{ Serial => LowSerial } +import scala.util.Try +import scala.concurrent._ + +class SerialOperator(serial: LowSerial, handler: ActorRef) extends Actor { + import context._ + + object Reader extends Thread { + private var continueReading = true + + override def run() { + while (continueReading) { + println("beginning read") + val data = ByteString(serial.read()) + println("return from read") + handler ! Received(data) + } + } + } + + Reader.start() + + context.watch(handler) + + def receive = { + case c @ Write(data) => { + val writer = sender + future{serial.write(data.toArray)}.onComplete { + case Success(data) => writer ! Wrote(ByteString(data)) + case Failure(t) => writer ! CommandFailed(c, t) + } + } + + case Close => { + serial.close() + sender ! Closed + context.stop(self) + } + } + + override def postStop = { + serial.close() + } + +} \ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/SerialOperator.scalac b/src/main/scala/com/github/jodersky/flow/SerialOperator.scalac deleted file mode 100644 index 21d2067..0000000 --- a/src/main/scala/com/github/jodersky/flow/SerialOperator.scalac +++ /dev/null @@ -1,44 +0,0 @@ -package com.github.jodersky.flow - -import scala.util.Failure -import scala.util.Success -import Serial._ -import akka.actor.Actor -import akka.actor.ActorLogging -import akka.actor.ActorRef -import akka.util.ByteString -import low.{ Serial => LowSerial } -import scala.util.Try -import scala.concurrent._ - -class SerialOperator(serial: LowSerial, handler: ActorRef) extends Actor { - import context._ - - context.watch(handler) - - class Reader extends Actor { - while (true) { - val data = ByteString(serial.read()) - handler ! Received(data) - } - } - - def receive = { - case Write(data) => { - val writer = sender - future{serial.write(data.toArray)}.onComplete { - case Success(data) => writer ! Wrote(ByteString(data)) - case Failure(t) => writer ! CommandFailed(c, t) - } - } - - case Close => { - context.stop(self) - } - } - - override def postStop = { - serial.close() - } - -} \ No newline at end of file -- cgit v1.2.3