wip - refactoring of RemoteWebcamWindow

Author: fagossa

build.sbt

@@ -8,6 +8,9 @@ libraryDependencies ++= Seq(
   "com.typesafe.akka" %% "akka-actor" % akkaVersion,
   "com.typesafe.akka" %% "akka-stream" % akkaVersion,
   "com.typesafe.akka" %% "akka-stream-contrib" % "0.4",
+  //"com.typesafe.akka" %% "akka-http-core" % akkaVersion,
+  "com.typesafe.akka" %% "akka-http-core" % "2.4.9",
+  "com.typesafe.akka" %% "akka-http-experimental" % "2.4.9",
   "org.typelevel" %% "cats" % "0.8.0",
   // mapping
   "com.typesafe.play" %% "play-json" % "2.5.8",

src/main/scala/fr/xebia/streams/WebcamWindow.scala src/main/scala/fr/xebia/streams/LocalWebcamWindow.scala

@@ -7,7 +7,7 @@ import fr.xebia.streams.transform.{ Flip, MediaConversion }
 import fr.xebia.streams.video.{ Dimensions, Webcam }
 import org.bytedeco.javacv.CanvasFrame
 
-object WebcamWindow extends App {
+object LocalWebcamWindow extends App {
 
   implicit val system = ActorSystem()
   implicit val materializer = ActorMaterializer()
@@ -17,9 +17,9 @@ object WebcamWindow extends App {
   canvas.setDefaultCloseOperation(javax.swing.JFrame.EXIT_ON_CLOSE)
 
   val imageDimensions = Dimensions(width = 640, height = 480)
-  val webcamSource = Webcam.local(deviceId = 0, dimensions = imageDimensions)
+  val localCameraSource = Webcam.local(deviceId = 0, dimensions = imageDimensions)
 
-  val graph = webcamSource
+  val graph = localCameraSource
     .map(MediaConversion.toMat) // most OpenCV manipulations require a Matrix
     .map(Flip.horizontal)
     .map(MediaConversion.toFrame) // convert back to a frame

src/main/scala/fr/xebia/streams/RemoteWebcamWindow.scala

@@ -0,0 +1,49 @@
+package fr.xebia.streams
+
+import akka.actor.ActorSystem
+import akka.stream.ActorMaterializer
+import akka.stream.scaladsl.{ Sink, Source }
+import akka.util.ByteString
+import fr.xebia.streams.transform.MediaConversion
+import fr.xebia.streams.video.Webcam
+import org.bytedeco.javacpp.opencv_core
+import org.bytedeco.javacpp.opencv_core.Mat
+import org.bytedeco.javacv.CanvasFrame
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.ExecutionContext
+
+object RemoteWebcamWindow extends App {
+
+  val logger = LoggerFactory.getLogger(getClass)
+
+  implicit val system = ActorSystem()
+  implicit val materializer = ActorMaterializer()
+
+  val canvas = new CanvasFrame("Webcam")
+  canvas.setDefaultCloseOperation(javax.swing.JFrame.EXIT_ON_CLOSE)
+
+  implicit val ec = system.dispatcher
+
+  val remoteCameraSource = Webcam.remote("192.168.0.17")
+
+  val graph = remoteCameraSource
+    .map(handleSource)
+    .map(_.map(MediaConversion.toFrame))
+    .map(_.map(canvas.showImage))
+    .map(_.to(Sink.ignore))
+
+  graph.map(_.run())
+
+  def handleSource(source: Source[ByteString, Any])(implicit ec: ExecutionContext) = {
+    source
+      .map { bytes => logger.info(bytes.toString()); bytes }
+      .map(_.toArray)
+      .map { bytes =>
+        val mat = new Mat(512, 288, opencv_core.CV_8SC3)
+        mat.data().put(bytes: _*)
+        mat
+      }
+  }
+
+}

src/main/scala/fr/xebia/streams/video/LocalCamFramePublisher.scala

@@ -21,7 +21,7 @@ private[video] class LocalCamFramePublisher(
   private implicit val ec = context.dispatcher
 
   // Lazy so that nothing happens until the flow begins
-  private lazy val grabber = buildGrabber(
+  private lazy val grabber: FrameGrabber = buildGrabber(
     deviceId = deviceId,
     imageWidth = imageWidth,
     imageHeight = imageHeight,

src/main/scala/fr/xebia/streams/video/RemoteCamFramePublisher.scala

@@ -0,0 +1,56 @@
+package fr.xebia.streams.video
+
+import akka.actor.{ ActorLogging, DeadLetterSuppression, Props }
+import akka.stream.actor.ActorPublisher
+import akka.stream.actor.ActorPublisherMessage.{ Cancel, Request }
+import akka.stream.scaladsl.Tcp
+import fr.xebia.streams.video.RemoteCamFramePublisher.{ Continue, buildGrabber }
+import org.bytedeco.javacv.Frame
+
+private[video] class RemoteCamFramePublisher(url: String)
+    extends ActorPublisher[Frame] with ActorLogging {
+
+  private implicit val ec = context.dispatcher
+
+  private lazy val grabber = buildGrabber(url)
+
+  def receive: Receive = {
+    case _: Request => emitFrames()
+    case Continue => emitFrames()
+    case Cancel => onCompleteThenStop()
+    case unexpectedMsg => log.warning(s"Unexpected message: $unexpectedMsg")
+  }
+
+  private def emitFrames(): Unit = {
+    if (isActive && totalDemand > 0) {
+      /*
+        Grabbing a frame is a blocking I/O operation, so we don't send too many at once.
+       */
+      grabFrame().foreach(onNext)
+      if (totalDemand > 0) {
+        self ! Continue
+      }
+    }
+  }
+
+  private def grabFrame(): Option[Frame] = {
+    Option(grabber.grab())
+  }
+
+}
+
+object RemoteCamFramePublisher {
+
+  def props(url: String): Props = Props(new RemoteCamFramePublisher(url))
+
+  def buildGrabber(url: String) = RemoteFrameGrabber(url)
+
+  private case object Continue extends DeadLetterSuppression
+
+  case class RemoteFrameGrabber(url: String) {
+    def grab(): Frame = ???
+  }
+
+  //Tcp().bindAndHandle()
+
+}

src/main/scala/fr/xebia/streams/video/Webcam.scala

@@ -1,36 +1,61 @@
 package fr.xebia.streams.video
 
 import akka.NotUsed
-import akka.actor.{ActorSystem, Props}
+import akka.actor.{ ActorSystem, Props }
+import akka.http.scaladsl.Http
+import akka.http.scaladsl.model.{ HttpRequest, HttpResponse }
+import akka.stream.Materializer
 import akka.stream.actor.ActorPublisher
-import akka.stream.scaladsl.Source
+import akka.stream.scaladsl.{ Framing, Source }
+import akka.util.{ ByteString, Timeout }
+import fr.xebia.streams.RemoteWebcamWindow._
 import org.bytedeco.javacpp.opencv_core._
 import org.bytedeco.javacv.Frame
 import org.bytedeco.javacv.FrameGrabber.ImageMode
+import org.slf4j.LoggerFactory
+
+import scala.concurrent.Future
 
 object Webcam {
 
   object local {
 
     def apply(
-               deviceId: Int,
-               dimensions: Dimensions,
-               bitsPerPixel: Int = CV_8U,
-               imageMode: ImageMode = ImageMode.COLOR
-             )(implicit system: ActorSystem): Source[Frame, NotUsed] = {
+      deviceId: Int,
+      dimensions: Dimensions,
+      bitsPerPixel: Int = CV_8U,
+      imageMode: ImageMode = ImageMode.COLOR
+    )(implicit system: ActorSystem): Source[Frame, NotUsed] = {
       val props: Props = LocalCamFramePublisher.props(deviceId, dimensions.width, dimensions.height, bitsPerPixel, imageMode)
       val webcamActorRef = system.actorOf(props)
-      val webcamActorPublisher = ActorPublisher[Frame](webcamActorRef)
+      val localActorPublisher = ActorPublisher[Frame](webcamActorRef)
 
-      Source.fromPublisher(webcamActorPublisher)
+      Source.fromPublisher(localActorPublisher)
     }
   }
 
   object remote {
 
-    def apply(host: String, port: String)
-             (implicit system: ActorSystem): Source[Frame, NotUsed] = {
-      Source.fromPublisher(???)
+    import scala.concurrent.duration._
+    val logger = LoggerFactory.getLogger(getClass)
+
+    implicit val timeout = Timeout(5.seconds)
+
+    val beginOfFrame = ByteString(0xff, 0xd8)
+
+    val endOfFrame = ByteString(0xff, 0xd9)
+
+    //http://doc.akka.io/docs/akka/2.4/scala/stream/stream-cookbook.html#chunking-up-a-stream-of-bytestrings-into-limited-size-bytestrings
+    def apply(host: String)(implicit system: ActorSystem, mat: Materializer): Future[Source[ByteString, Any]] = {
+      implicit val ec = system.dispatcher
+      val httpRequest = HttpRequest(uri = "/html/cam_pic_new.php")
+
+      Http()
+        .singleRequest(httpRequest)
+        .map { response => logger.info(response.toString()); response }
+        .map(_.entity.dataBytes)
+        .map(_.via(Framing.delimiter(endOfFrame, maximumFrameLength = 100, allowTruncation = true)))
+        .map(_.map(_.dropWhile(_ != beginOfFrame)))
     }
 
   }