How to bind akka http with akka streams?

Question

I'm trying to use streams instead of pure actors to handle http requests and I came with the following code:

trait ImagesRoute {

  val log = LoggerFactory.getLogger(this.getClass)

  implicit def actorRefFactory: ActorRefFactory
  implicit def materializer: ActorMaterializer

  val source =
    Source
      .actorRef[Image](Int.MaxValue, OverflowStrategy.fail)
      .via(Flow[Image].mapAsync(1)(ImageRepository.add))
      .toMat(Sink.asPublisher(true))(Keep.both)

  val route = {
    pathPrefix("images") {
      pathEnd {
        post {
          entity(as[Image]) { image =>

            val (ref, publisher) = source.run()

            val addFuture = Source.fromPublisher(publisher)

            val future = addFuture.runWith(Sink.head[Option[Image]])

            ref ! image

            onComplete(future.mapTo[Option[Image]]) {
              case Success(img) =>
                complete(Created, img)

              case Failure(e) =>
                log.error("Error adding image resource", e)
                complete(InternalServerError, e.getMessage)
            }
          }
        }
      }
    }
  }
}

I'm not sure if this is the correct way to do that, or even if this is a good approach or if I should use an actor to interact with the route, using the ask pattern and then inside the actor, stream everything.

Any ideas?

If I'm not mistaken, in your case you don't need streams at all. As far as I understand, the `ImageRepository.add` method returns a `Future`; all you need to do then is to write `onComplete(ImageRepository.add(image))`, and that's all. — Vladimir Matveev, Sep 21 '16 at 19:26
@VladimirMatveev yes that's right, this is just a simple example, but the stream pipeline should be bigger doing a lot of things like contacting external resources and eventually back pressure things... — Thiago Pereira, Sep 21 '16 at 19:32

score 5 · Accepted Answer · edited May 23 '17 at 12:32

If you're only expecting 1 image from the entity then you don't need to create a Source from an ActorRef and you don't need Sink.asPublisher, you can simply use Source.single:

def imageToComplete(img : Option[Image]) : StandardRoute = 
  img.map(i => complete(Created, i))
     .getOrElse {
       log error ("Error adding image resource", e)
       complete(InternalServerError, e.getMessage
     }

...

entity(as[Image]) { image =>

  val future : Future[StandardRoute] = 
    Source.single(image)
          .via(Flow[Image].mapAsync(1)(ImageRepository.add))
          .runWith(Sink.head[Option[Image]])
          .map(imageToComplete)

  onComplete(future)
}

Simplifying your code further, the fact that you are only processing 1 image means that Streams are unnecessary since there is no need for backpressure with just 1 element:

val future : Future[StandardRoute] = ImageRepository.add(image)
                                                    .map(imageToComplete)

onComplete(future)

In the comments you indicated

"this is just a simple example, but the stream pipeline should be bigger doing a lot of things like contacting external resources and eventually back pressure things"

This would only apply if your entity was a stream of images. If you're only ever processing 1 image per HttpRequest then backpressure never applies, and any stream you create will be a slower version of a Future.

If your entity is in fact a stream of Images, then you could use it as part of stream:

val byteStrToImage : Flow[ByteString, Image, _] = ???

val imageToByteStr : Flow[Image, Source[ByteString], _] = ???

def imageOptToSource(img : Option[Image]) : Source[Image,_] =
  Source fromIterator img.toIterator

val route = path("images") {
  post {
    extractRequestEntity { reqEntity =>

      val stream = reqEntity.via(byteStrToImage)
                            .via(Flow[Image].mapAsync(1)(ImageRepository.add))
                            .via(Flow.flatMapConcat(imageOptToSource))
                            .via(Flow.flatMapConcat(imageToByteStr))

      complete(HttpResponse(status=Created,entity = stream))
    }
  }
}

Thank you, that makes a lot of sense – Thiago Pereira Sep 22 '16 at 13:45 — Thiago Pereira, Sep 22 '16 at 13:45

How to bind akka http with akka streams?

1 Answers1