How to backpressure a ActorPublisher

Question

I'm writing few samples to understand akka streams and backpressures. I'm trying to see how a slow consumer backpressure's a AkkaPublisher

My code as follows.

class DataPublisher extends ActorPublisher[Int] {

  import akka.stream.actor.ActorPublisherMessage._

  var items: List[Int] = List.empty

  def receive = {
    case s: String =>
      println(s"Producer buffer size ${items.size}")
      if (totalDemand == 0)
        items = items :+ s.toInt
      else
        onNext(s.toInt)

    case Request(demand) =>
      if (demand > items.size) {
        items foreach (onNext)
        items = List.empty
      }
      else {
        val (send, keep) = items.splitAt(demand.toInt)
        items = keep
        send foreach (onNext)
      }


    case other =>
      println(s"got other $other")
  }
}

and

Source.fromPublisher(ActorPublisher[Int](dataPublisherRef)).runWith(sink)

Where the sink is a Subscriber with a sleep to emulate slow consumer. And publisher keeps producing data regardless.

--EDIT-- My question is when the demand is 0 programatically buffers data. How can I make use of backpressure to slow down the publisher

Something like

throttledSource().buffer(10, OverflowStrategy.backpressure).runWith(throttledSink())

This will not effect the publisher and its buffer keeps going.

Thanks, Sajith

Answer 1

Don't use ActorPublisher

Firstly, don't use ActorPublisher - it is a very low-level and deprecated API. We decided to deprecate as users should not be working on such low level of abstraction in Akka Streams.

One of the tricky things is exactly what you're asking about -- handling backpressure is entirely in the hands of the developer writing the ActorPublisher if they use this API. So you have to receive the Request(n) messages and make sure that you never signal more elements than you got requests for. This behaviour is specified in the Reactive Streams Specification which you then have to implement correctly. Basically, you're exposed to all the complexities of Reactive Streams (which is a full specification, with many edge cases -- disclaimer: I was/am part of developing Reactive Streams as well as Akka Streams).

Showing how back-pressure manifests in GraphStage

Secondly, to build custom stages you should be using the API designed for it: GraphStage. Please note that such stage is also pretty low level. Normally users of Akka Streams don't need to write custom stages, however it is absolutely expected and fine to write your own stages if they would implement some logic that the built-in stages don't provide.

Here's a simplified Filter implementation from the Akka codebase:

case class Filter[T](p: T ⇒ Boolean) extends SimpleLinearGraphStage[T] {
  override def initialAttributes: Attributes = DefaultAttributes.filter

  override def toString: String = "Filter"

  override def createLogic(inheritedAttributes: Attributes): GraphStageLogic =
    new GraphStageLogic(shape) with OutHandler with InHandler {

    override def onPush(): Unit = {
      val elem = grab(in)
      if (p(elem)) push(out, elem)
      else pull(in)
    }

    // this method will NOT be called, if the downstream has not signalled enough demand!
    // this method being NOT called is how back-pressure manifests in stages
    override def onPull(): Unit = pull(in)

    setHandlers(in, out, this)
  }
}

As you can see, instead of implementing the entire Reactive Streams logic and rules yourself (which is hard), you get simple callbacks like onPush and onPull. Akka Streams handles the demand management, and it will automatically call onPull if the downstream has signaled demand, and it will NOT call it, if there is no demand -- which would mean the downstream is applying backpressure to this stage.