diff options
author | Jakob Odersky <jakob@odersky.com> | 2016-02-01 22:46:48 -0800 |
---|---|---|
committer | Jakob Odersky <jakob@odersky.com> | 2016-02-01 22:46:48 -0800 |
commit | 50bc6c80aa201c0543db07a62bc7c726b813c168 (patch) | |
tree | 0ed1c753f9d9fe29d1e51a0d03172e5659c4c339 /mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala | |
parent | 4f4c799a6d9ccf333a3e609a2464e2f317875af7 (diff) | |
download | mavigator-50bc6c80aa201c0543db07a62bc7c726b813c168.tar.gz mavigator-50bc6c80aa201c0543db07a62bc7c726b813c168.tar.bz2 mavigator-50bc6c80aa201c0543db07a62bc7c726b813c168.zip |
Use Akka streams and temporarily disable frontend
Diffstat (limited to 'mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala')
-rw-r--r-- | mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala | 20 |
1 files changed, 20 insertions, 0 deletions
diff --git a/mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala b/mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala new file mode 100644 index 0000000..5e48ea1 --- /dev/null +++ b/mavigator-uav/src/main/scala/mavigator/uav/Multiplexer.scala @@ -0,0 +1,20 @@ +package mavigator +package uav + +import akka.stream.scaladsl._ +import akka.stream._ +import org.reactivestreams._ + +private[uav] class Multiplexer[In, Out](service: Flow[In, Out, _])(implicit materializer: Materializer) { + + private val endpoint: Flow[Out, In, (Publisher[Out], Subscriber[In])] = Flow.fromSinkAndSourceMat( + Sink.asPublisher[Out](fanout = true), + Source.asSubscriber[In])((pub, sub) => (pub, sub)) + + private lazy val (publisher, subscriber) = (service.joinMat(endpoint)(Keep.right)).run() + + def connect(client: Flow[Out, In, _]) = { + Source.fromPublisher(publisher).via(client).to(Sink.ignore).run() + } + +} |