# Backpressure with Actors **URL:** \ **Category:** Akka Libraries\ **Tags:** akka-typed, scala, streams\ **Created:** [April 23, 2024, 5:11am UTC](https://discuss.akka.io/t/backpressure-with-actors/10710 "2024-04-23T05:11:49Z")\ **Posts on this page:** 12\ **Page:** 1
**Author:** ![sosborn-trumid](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/sosborn-trumid/32/2979_2.png) [@sosborn-trumid](https://discuss.akka.io/u/sosborn-trumid)\ **Post date:** [April 23, 2024, 5:11am UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/1 "2024-04-23T05:11:49Z")
Hi there, I am having trouble wrapping my head around the various actorRefWithBackpressure options. Let’s say I have the following: 1. A fast source 2. An actor that consumes and processes that source but is somewhat slow. Is the following the correct way to hook these up? And if so, where am I supposed to be emitting and/or handling the Ack message? Is this somehow wired up by this construction: fastSource.runWith(ActorRef.actorRefWithBackpressure(slowActor???, Ack, …)) I have read all the online docs, but they seem to be a few steps short of what I need to really grok what is going on, so any more extensive examples would be much appreciated. Thanks. ---
**Author:** ![Synduex](https://avatars.discourse-cdn.com/v4/letter/s/f475e1/32.png) [@Synduex](https://discuss.akka.io/u/Synduex)\ **Post date:** [April 23, 2024, 4:43pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/2 "2024-04-23T16:43:29Z")
Hi @sosborn-trumid, Can I ask for the use case. In Akka Streams Source converts the data into streams, flow transforms that data and allows you to materialize it and Sink is the consumer of that data. If you want to lets say send data from a source to an Actor. You can use Tell (!) pattern to achieve it as Sink is consuming it. In Sink you can apply foreach and send each data unit as a message. [Sink.foreach[Int]( x =\> Actor ! newNumber(x)) If you want to ensure that the data is sent to the Sink only when its available - check out Conflate in Flow that acts like fold operator but emits the result only when downstream is ready. ---
**Author:** ![sosborn-trumid](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/sosborn-trumid/32/2979_2.png) [@sosborn-trumid](https://discuss.akka.io/u/sosborn-trumid)\ **Post date:** [April 23, 2024, 4:57pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/3 "2024-04-23T16:57:04Z")
Thanks, yes. I was going down the conflate route, but I still could not see how to trigger conflating because it needs some sort of backpressure signal to kick in, and where is that coming from, since the consumer is an Actor that just has a mailbox accepting all messages. My use case is a little more complicated. It is Source → Snapshotting Actor → Processing Actor, where the processing actor is the potentially slow bit. But at this point, I’m just trying to understand the mechanics of applying backpressure from an actor to a source and finding the documentation somewhat confusing. `ActorSink.actorRefWithBackpressure` seems like the right bridge, but I’m still unclear how it all goes together and whether or not I need to be sending Ack messages to someone. If I have SlowActor[M] where M is my `trait MyMessage`, it seems like I need to make the protocol messages (Init, etc.) implement MyMessage. Is that correct? ---
**Author:** ![Synduex](https://avatars.discourse-cdn.com/v4/letter/s/f475e1/32.png) [@Synduex](https://discuss.akka.io/u/Synduex)\ **Post date:** [April 23, 2024, 5:25pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/4 "2024-04-23T17:25:09Z")
I see try this sink example. import akka.stream.typed.scaladsl.ActorSink val sink = ActorSink.actorRefWithBackpressure( ref = myActor, onInitMessage = (actorRef: ActorRef[String]) =\> “Stream started”, ackMessage = “Ack”, onCompleteMessage = “Stream completed”, onFailureMessage = (ex: Throwable) =\> ex.getMessage ) source.runWith(sink) Replace ackMessage with the acknowledge message . Hope this helps! ---
**Author:** ![Synduex](https://avatars.discourse-cdn.com/v4/letter/s/f475e1/32.png) [@Synduex](https://discuss.akka.io/u/Synduex)\ **Post date:** [April 23, 2024, 5:33pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/5 "2024-04-23T17:33:16Z")
@sosborn-trumid Yes, you’re on the right track. The ActorSink.actorRefWithBackpressure method is designed to handle situations like this, where you have a fast producer (source) and a slower consumer (actor in this case). ---
**Author:** ![Synduex](https://avatars.discourse-cdn.com/v4/letter/s/f475e1/32.png) [@Synduex](https://discuss.akka.io/u/Synduex)\ **Post date:** [April 23, 2024, 5:37pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/6 "2024-04-23T17:37:56Z")
Here is a demo actor // Define the messages case class StreamElement(value: Int) case object Ack // Define the actor class SlowActor extends Actor { def receive: Receive = { case StreamElement(value) =\> println(s"Processing $value") Thread.sleep(1000) // Simulate slow processing sender() ! Ack // Send acknowledgement back to the stream } } ---
**Author:** ![leviramsey](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/leviramsey/32/2653_2.png) [@leviramsey](https://discuss.akka.io/u/leviramsey)\ **Post date:** [April 23, 2024, 5:59pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/7 "2024-04-23T17:59:45Z")
For an `ActorSink.actorRefWithBackpressure`, the type parameters are `T`, `M`, and `A`. In this case, `M` will be `MyMessage`, `T` is whatever type your stream is feeding into the sink, and `A` is the ack-message, which signals the sink to stop backpressuring (viz. that the actor the sink is sending to is ready to handle a new message). So whatever message is constructed as the init-message, the complete-message, and the failure-message (as well as whatever message is constructed in the `messageBuilder`) would have to be a `MyMessage`. The ack-message does not have be a `MyMessage` (`A` can be literally any type): the only effective requirement (not captured in the type system, alas) is that whatever message is constructed in `messageBuilder` has to have a field of type `A`. A reasonable convention to follow in the absence of any further requirement is that this field will be named `replyTo` and have the type `akka.Done`. ---
**Author:** ![sosborn-trumid](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/sosborn-trumid/32/2979_2.png) [@sosborn-trumid](https://discuss.akka.io/u/sosborn-trumid)\ **Post date:** [April 23, 2024, 9:20pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/8 "2024-04-23T21:20:09Z")
Thank you both - this has been super helpful. I think I’ve got a handle on it now. Cheers! ---
**Author:** ![Synduex](https://avatars.discourse-cdn.com/v4/letter/s/f475e1/32.png) [@Synduex](https://discuss.akka.io/u/Synduex)\ **Post date:** [April 23, 2024, 9:21pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/9 "2024-04-23T21:21:42Z")
Cheers! Happy coding! ---
**Author:** ![sosborn-trumid](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/sosborn-trumid/32/2979_2.png) [@sosborn-trumid](https://discuss.akka.io/u/sosborn-trumid)\ **Post date:** [April 24, 2024, 11:14pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/10 "2024-04-24T23:14:02Z")
Couple more small follow-up items. 1. Am I supposed to send an Ack in response to the Init message, or just in response to the forwarded Messages? 2. I have noticed, at least in my test scenerios, that I can sometimes receive the first Message _before_ the Init message. Is that supposed to be able to happen? ---
**Author:** ![johanandren](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/johanandren/32/85_2.png) [@johanandren](https://discuss.akka.io/u/johanandren)\ **Post date:** [April 25, 2024, 6:53am UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/11 "2024-04-25T06:53:25Z")
The stream will not start passing actual elements until it gets an ack for the init message. If you get messages before that it would be a bug, if you can make a minimal reproducer we’d be interested in seeing that (you can see the test coverage in the Akka test suite for that specific case [here](https://github.com/akka/akka/blob/5e9e11b5f6df03ed89c5ca37bdbb295ff5c19ec4/akka-stream-tests/src/test/scala/akka/stream/scaladsl/ActorRefBackpressureSinkSpec.scala#L108)). ---
**Author:** ![sosborn-trumid](https://sea2.discourse-cdn.com/flex020/user_avatar/discuss.akka.io/sosborn-trumid/32/2979_2.png) [@sosborn-trumid](https://discuss.akka.io/u/sosborn-trumid)\ **Post date:** [April 25, 2024, 11:26pm UTC](https://discuss.akka.io/t/backpressure-with-actors/10710/12 "2024-04-25T23:26:24Z")
Thank you. Turns out the the stray message I was seeing was coming from somewhere else, so all is right with the world, and it behaves as you say.