diff options
author | Jakob Odersky <jodersky@gmail.com> | 2013-06-18 19:40:22 +0200 |
---|---|---|
committer | Jakob Odersky <jodersky@gmail.com> | 2013-06-18 19:40:22 +0200 |
commit | 98241aa830bedb006ae041dce661afd57c3d90a8 (patch) | |
tree | 70443e7597aca759733e587220a8323fde89af1f /src/main | |
parent | b0c32f5325702dd7f7ef3d5ccc0eb9a2b972cf7a (diff) | |
download | akka-serial-98241aa830bedb006ae041dce661afd57c3d90a8.tar.gz akka-serial-98241aa830bedb006ae041dce661afd57c3d90a8.tar.bz2 akka-serial-98241aa830bedb006ae041dce661afd57c3d90a8.zip |
use sbt-native plugin and restructure build
Diffstat (limited to 'src/main')
10 files changed, 720 insertions, 0 deletions
diff --git a/src/main/java/com/github/jodersky/flow/low/NativeSerial.java b/src/main/java/com/github/jodersky/flow/low/NativeSerial.java new file mode 100644 index 0000000..6bdcde5 --- /dev/null +++ b/src/main/java/com/github/jodersky/flow/low/NativeSerial.java @@ -0,0 +1,49 @@ +package com.github.jodersky.flow.low; + +class NativeSerial { + + static { + NativeLoader.load(); + } + + final static int E_PERMISSION = -1; + final static int E_OPEN = -2; + final static int E_BUSY = -3; + final static int E_BAUD = -4; + final static int E_PIPE = -5; + final static int E_MALLOC = -6; + final static int E_POINTER = -7; + final static int E_POLL = -8; + final static int E_IO = -9; + final static int E_CLOSE = -10; + + + /* return values: + * 0 ok + * E_PERMISSION don't have permission to open + * E_OPEN can't get file descriptor + * E_BUSY device busy + * E_BAUD invalid baudrate + * E_PIPE can't open pipe for graceful closing + * E_MALLOC malloc error */ + native static int open(String device, int baud, long[] serial); + + /* return + * >0 number of bytes read + * E_POINTER invalid serial pointer + * E_POLL poll error + * E_IO read error + * E_CLOSE close request */ + native static int read(long serial, byte[] buffer); + + /*return + * >0 number of bytes written + * E_POINTER invalid serial config (null pointer) + * E_IO write error */ + native static int write(long serial, byte[] buffer); + + native static void close(long serial); + + native static void debug(boolean value); + +} diff --git a/src/main/native/flow.c b/src/main/native/flow.c new file mode 100644 index 0000000..09ffca6 --- /dev/null +++ b/src/main/native/flow.c @@ -0,0 +1,294 @@ +/* + * Copyright (C) 2013 Jakob Odersky + * + * Redistribution and use in source and binary forms, with or without + * modification, are permitted provided that the following conditions are + * met: + * + * * Redistributions of source code must retain the above copyright + * notice, this list of conditions and the following disclaimer. + * * Redistributions in binary form must reproduce the above + * copyright notice, this list of conditions and the following disclaimer + * in the documentation and/or other materials provided with the + * distribution. + * * Neither the name of the nor the names of its + * contributors may be used to endorse or promote products derived from + * this software without specific prior written permission. + * + * THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS + * "AS IS" AND ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT + * LIMITED TO, THE IMPLIED WARRANTIES OF MERCHANTABILITY AND FITNESS FOR + * A PARTICULAR PURPOSE ARE DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT + * OWNER OR CONTRIBUTORS BE LIABLE FOR ANY DIRECT, INDIRECT, INCIDENTAL, + * SPECIAL, EXEMPLARY, OR CONSEQUENTIAL DAMAGES (INCLUDING, BUT NOT + * LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR SERVICES; LOSS OF USE, + * DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER CAUSED AND ON ANY + * THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY, OR TORT + * (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE + * OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE. + * + */ + +#include <stdlib.h> +#include <stdio.h> +#include <stdbool.h> +#include <unistd.h> +#include <errno.h> +#include <termios.h> +#include <fcntl.h> +#include <poll.h> +#include "com_github_jodersky_flow_low_NativeSerial.h" + +#define E_PERMISSION -1 +#define E_OPEN -2 +#define E_BUSY -3 +#define E_BAUD -4 +#define E_PIPE -5 +#define E_MALLOC -6 +#define E_POINTER -7 +#define E_POLL -8 +#define E_IO -9 +#define E_CLOSE -10 + + +static bool debug = false; +#define DEBUG(f) if (debug) {f;} + +//contains file descriptors used in managing a serial port +struct serial_config { + + int fd; //serial port + int pipe_read; //event + int pipe_write; //event + +}; + +/* return values: + * 0 ok + * E_PERMISSION don't have permission to open + * E_OPEN can't get file descriptor + * E_BUSY device busy + * E_BAUD invalid baudrate + * E_PIPE can't open pipe for graceful closing + * E_MALLOC malloc error + */ +int serial_open(const char* device, int baud, struct serial_config** serial) { + + int fd = open(device, O_RDWR | O_NOCTTY | O_NONBLOCK); + + if (fd < 0) { + DEBUG(perror(device)); + if (errno == EACCES) return E_PERMISSION; + else return E_OPEN; + } + + if (flock(fd, LOCK_EX | LOCK_NB) < 0) { + DEBUG(perror(device)); + return E_BUSY; + } + + speed_t bd; + switch (baud) { + case 50: bd = B50; break; + case 75: bd = B75; break; + case 110: bd = B110; break; + case 134: bd = B134; break; + case 150: bd = B150; break; + case 200: bd = B200; break; + case 300: bd = B300; break; + case 600: bd = B600; break; + case 1200: bd = B1200; break; + case 1800: bd = B1800; break; + case 2400: bd = B2400; break; + case 4800: bd = B4800; break; + case 9600: bd = B9600; break; + case 19200: bd = B19200; break; + case 38400: bd = B38400; break; + case 57600: bd = B57600; break; + case 115200: bd = B115200; break; + case 230400: bd = B230400; break; + default: return E_BAUD; break; + } + + /* configure new port settings */ + struct termios newtio; + newtio.c_cflag &= ~(PARENB | CSTOPB | CSIZE | CRTSCTS); // 8N1 + newtio.c_cflag |= CS8 | CREAD | CLOCAL; + newtio.c_iflag &= ~(IXON | IXOFF | IXANY); // turn off s/w flow ctrl + newtio.c_lflag &= ~(ICANON | ECHO | ECHOE | ISIG); // make raw + newtio.c_oflag &= ~OPOST; // make raw + + // see: http://unixwiz.net/techtips/termios-vmin-vtime.html + //newtio.c_cc[VMIN] = 1; + //newtio.c_cc[VTIME] = 2*10/baud; + cfsetspeed(&newtio, bd); + + /* load new settings to port */ + tcflush(fd, TCIOFLUSH); + tcsetattr(fd,TCSANOW,&newtio); + + int pipe_fd[2]; + if (pipe2(pipe_fd, O_NONBLOCK) < 0) { + DEBUG(perror(device)); + return E_PIPE; + } + + struct serial_config* s = malloc(sizeof(s)); + if (s == NULL) { + return E_MALLOC; + } + + s->fd = fd; + s->pipe_read = pipe_fd[0]; + s->pipe_write = pipe_fd[1]; + (*serial) = s; + + return 0; +} + +void serial_close(struct serial_config* serial) { + + if (serial == NULL) return; + + int data = 0xffffffff; + + //write to pipe to wake up any blocked read thread (self-pipe trick) + if (write(serial->pipe_write, &data, 1) <= 0) { + DEBUG(perror("error writing to pipe during close")) + } + + close(serial->pipe_write); + close(serial->pipe_read); + + flock(serial->fd, LOCK_UN); + close(serial->fd); + + free(serial); +} + +/* return + * >0 number of bytes read + * E_POINTER invalid serial pointer + * E_POLL poll error + * E_IO read error + * E_CLOSE close request + */ +int serial_read(struct serial_config* serial, unsigned char * buffer, size_t size) { + if (serial == NULL) { + return E_POINTER; + } + + struct pollfd sp; //serial poll + sp.fd = serial->fd; + sp.events = POLLIN; + + struct pollfd pp; //pipe poll + pp.fd = serial->pipe_read; + pp.events = POLLIN; + + struct pollfd poll_list[] = {sp, pp}; + + int n = poll(poll_list,(unsigned long)3,-1); + if (n < 0) { + DEBUG(perror("read")); + return E_IO; + } + + if (sp.revents & POLLIN != 0) { + int r = read(sp.fd, buffer, size); + + //treat 0 bytes read as an error to avoid problems on disconnect + //anyway, after a poll there should be more than 0 bytes available to read + if (r <= 0) { + if (r < 0) DEBUG(perror("read")); + return E_IO; + } + return r; + } else { + return E_CLOSE; + } +} + +/*return + * >0 number of bytes written + * E_POINTER invalid serial config (null pointer) + * E_IO write error + */ +int serial_write(struct serial_config* serial, unsigned char* data, size_t size) { + if (serial == NULL) return E_POINTER; + + int r = write(serial->fd, data, size); + if (r < 0) { + DEBUG(perror("write")); + return E_IO; + } + return r; +} + + + +// JNI bindings +// ============ + +inline struct serial_config* j2s(jlong pointer) { + return (struct serial_config*) pointer; +} + +inline jlong s2j(struct serial_config* pointer) { + return (jlong) pointer; +} + +JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_open + (JNIEnv *env, jclass clazz, jstring device, jint baud, jlongArray jserialp) +{ + const char *dev = (*env)->GetStringUTFChars(env, device, 0); + struct serial_config* serial; + int r = serial_open(dev, baud, &serial); + (*env)->ReleaseStringUTFChars(env, device, dev); + + long serialp = s2j(serial); + (*env)->SetLongArrayRegion(env, jserialp, 0, 1, &serialp); + + return r; +} + +JNIEXPORT void JNICALL Java_com_github_jodersky_flow_low_NativeSerial_close + (JNIEnv * env, jclass clazz, jlong serial) +{ + serial_close(j2s(serial)); +} + +JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_read + (JNIEnv * env, jclass clazz, jlong serial, jbyteArray jbuffer) +{ + + jsize size = (*env)->GetArrayLength(env, jbuffer); + + unsigned char buffer[size]; + int n = serial_read(j2s(serial), buffer, size); + if (n < 0) { + return n; + } + + (*env)->SetByteArrayRegion(env, jbuffer, 0, n, (signed char *) buffer); + return n; +} + +JNIEXPORT jint JNICALL Java_com_github_jodersky_flow_low_NativeSerial_write + (JNIEnv * env, jclass clazz, jlong serial, jbyteArray jbuffer) +{ + unsigned char * buffer = (*env)->GetByteArrayElements(env, jbuffer, NULL); + int size = (*env)->GetArrayLength(env, jbuffer); + int r = serial_write(j2s(serial), buffer, size); + + (*env)->ReleaseByteArrayElements(env, jbuffer, buffer, JNI_ABORT); + + return r; +} + +JNIEXPORT void JNICALL Java_com_github_jodersky_flow_low_NativeSerial_debug + (JNIEnv *env, jclass clazz, jboolean value) +{ + debug = (bool) value; +} + diff --git a/src/main/scala/com/github/jodersky/flow/Framing.scalac b/src/main/scala/com/github/jodersky/flow/Framing.scalac new file mode 100644 index 0000000..f8173a7 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/Framing.scalac @@ -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/Serial.scalac b/src/main/scala/com/github/jodersky/flow/Serial.scalac new file mode 100644 index 0000000..7182425 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/Serial.scalac @@ -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 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.scalac b/src/main/scala/com/github/jodersky/flow/SerialManager.scalac new file mode 100644 index 0000000..ca3fc6b --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/SerialManager.scalac @@ -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) => 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.scalac b/src/main/scala/com/github/jodersky/flow/SerialOperator.scalac new file mode 100644 index 0000000..21d2067 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/SerialOperator.scalac @@ -0,0 +1,44 @@ +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 diff --git a/src/main/scala/com/github/jodersky/flow/exceptions.scala b/src/main/scala/com/github/jodersky/flow/exceptions.scala new file mode 100644 index 0000000..c7b2fa1 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/exceptions.scala @@ -0,0 +1,8 @@ +package com.github.jodersky.flow + +import java.io.IOException + +class NoSuchPortException(message: String) extends IOException(message) +class PortInUseException(message: String) extends IOException(message) +class AccessDeniedException(message: String) extends IOException(message) +class PortClosingException(message: String) extends IOException(message)
\ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/low/NativeLoader.scala b/src/main/scala/com/github/jodersky/flow/low/NativeLoader.scala new file mode 100644 index 0000000..fda82a6 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/low/NativeLoader.scala @@ -0,0 +1,35 @@ +package com.github.jodersky.flow.low + +import java.io.File +import java.io.FileOutputStream + +object NativeLoader { + + def load = { + val os = System.getProperty("os.name").toLowerCase + val arch = System.getProperty("os.arch").toLowerCase + + val in = NativeLoader.getClass().getResourceAsStream("/native/" + os + "/" + arch + "/" + "libflow.so") + val temp = File.createTempFile("flow" + os + arch, ".so"); + temp.deleteOnExit() + val out = new FileOutputStream(temp); + + try { + var read: Int = 0; ; + val buffer = new Array[Byte](4096); + do { + read = in.read(buffer) + if (read != -1) { + out.write(buffer, 0, read); + } + } while (read != -1) + } finally { + in.close() + out.close + } + + System.load(temp.getAbsolutePath()) + + } + +}
\ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/low/Serial.scala b/src/main/scala/com/github/jodersky/flow/low/Serial.scala new file mode 100644 index 0000000..e482bf8 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/low/Serial.scala @@ -0,0 +1,65 @@ +package com.github.jodersky.flow.low + +import scala.concurrent._ +import scala.concurrent.ExecutionContext.Implicits._ +import java.io.IOException +import com.github.jodersky.flow.AccessDeniedException +import com.github.jodersky.flow.NoSuchPortException +import com.github.jodersky.flow.PortInUseException +import com.github.jodersky.flow.PortClosingException +import scala.util.Try + +class Serial private (val port: String, private val pointer: Long) { + import NativeSerial._ + + def read(): Array[Byte] = synchronized { + val buffer = new Array[Byte](100) + NativeSerial.read(pointer, buffer) match { + case E_POINTER => throw new NullPointerException("pointer to native serial") + case E_POLL => throw new IOException(port + ": polling") + case E_IO => throw new IOException(port + ": reading") + case E_CLOSE => throw new PortClosingException(port + " closing") + case bytes if bytes > 0 => buffer.take(bytes) + case error => throw new IOException(s"unknown read error ${error}") + } + } + + def write(data: Array[Byte]): Array[Byte] = { + import NativeSerial._ + NativeSerial.write(pointer, data) match { + case E_POINTER => throw new NullPointerException("pointer to native serial") + case E_IO => throw new IOException(port + ": writing") + case bytes if bytes > 0 => data.take(bytes) + case error => throw new IOException(s"unknown write error ${error}") + } + } + + def close() = { + NativeSerial.close(pointer) + } + +} + +object Serial { + + def open(port: String, baud: Int) = synchronized { + val pointer = new Array[Long](1) + val result = NativeSerial.open(port, baud, pointer) + + import NativeSerial._ + + result match { + case E_PERMISSION => throw new AccessDeniedException(port) + case E_OPEN => throw new NoSuchPortException(port) + case E_BUSY => throw new PortInUseException(port) + case E_BAUD => throw new IllegalArgumentException(s"invalid baudrate ${baud}, use standard unix values") + case E_PIPE => throw new IOException("cannot create pipe") + case E_MALLOC => throw new IOException("cannot allocate memory for serial port") + case 0 => new Serial(port, pointer(0)) + case error => throw new IOException(s"unknown error ${error}") + } + } + + def debug(value: Boolean) = NativeSerial.debug(value) + +}
\ No newline at end of file diff --git a/src/main/scala/com/github/jodersky/flow/test.sc b/src/main/scala/com/github/jodersky/flow/test.sc new file mode 100644 index 0000000..88a2193 --- /dev/null +++ b/src/main/scala/com/github/jodersky/flow/test.sc @@ -0,0 +1,52 @@ +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 +import akka.io._ +import akka.actor.{IO=>_,_} +import Serial._ + +object test { + + val StartByte = 0: Byte //> StartByte : Byte = 0 + val StopByte = 1: Byte //> StopByte : Byte = 1 + val EscapeByte = 2: Byte //> EscapeByte : Byte = 2 + + val ctx = new PipelineContext{} //> ctx : akka.io.PipelineContext = com.github.jodersky.flow.test$$anonfun$main + //| $1$$anon$1@32bf7190 + + val stages = new DelimitedFrame(StartByte, StopByte, EscapeByte) + //> stages : com.github.jodersky.flow.DelimitedFrame = com.github.jodersky.flow + //| .DelimitedFrame@36ff057f + + val PipelinePorts(cmd, evt, mgmt) = PipelineFactory.buildFunctionTriple(ctx, stages) + //> cmd : akka.util.ByteString => (Iterable[akka.util.ByteString], Iterable[akk + //| a.util.ByteString]) = <function1> + //| evt : akka.util.ByteString => (Iterable[akka.util.ByteString], Iterable[akk + //| a.util.ByteString]) = <function1> + //| mgmt : PartialFunction[AnyRef,(Iterable[akka.util.ByteString], Iterable[akk + //| a.util.ByteString])] = <function1> + + val injector = PipelineFactory.buildWithSinkFunctions(ctx, stages)( + t => println("sent command: " + t), // will receive messages of type Try[ByteString] + t => println("got event: " + t) // will receive messages of type Try[Message] + ) //> injector : akka.io.PipelineInjector[akka.util.ByteString,akka.util.ByteStri + //| ng] = akka.io.PipelineFactory$$anon$5@70cb6009 + + + val bs = ByteString.fromArray(Array(0,4,2,1,1,6,1)) + //> bs : akka.util.ByteString = ByteString(0, 4, 2, 1, 1, 6, 1) + + injector.injectCommand(bs) //> sent command: Success(ByteString(0, 2, 0, 4, 2, 2, 2, 1, 2, 1, 6, 2, 1, 1)) + injector.injectEvent(bs) //> got event: Success(ByteString(4, 1)) + + implicit val system = ActorSystem("flow") + //> system : akka.actor.ActorSystem = akka://flow| + //IO(Serial) ! Open("s", 9600) + +}
\ No newline at end of file |