You signed in with another tab or window. Reload to refresh your session.You signed out in another tab or window. Reload to refresh your session.You switched accounts on another tab or window. Reload to refresh your session.Dismiss alert
Hi
When I send many ChannelMessage with a high rate to ChannelActor with dropIfNoChannel = false if I stop rabbitmq and after a duration of time like 1 minute start rabbitmq again, the ChannelActor blocked for process many messages and can't process other message and throw heap space error.
I write this code:
import akka.actor.{ActorRef, ActorSystem}
import akka.pattern._
import akka.util.Timeout
import com.newmotion.akka.rabbitmq.{ChannelActor, ChannelCreated, ChannelMessage, ConnectionActor, ConnectionFactory, CreateChannel}
import com.rabbitmq.client.MessageProperties
import scala.annotation.tailrec
import scala.collection.mutable
import scala.concurrent.ExecutionContext
import scala.concurrent.duration._
object Exam1 extends App {
val system = ActorSystem()
implicit val timeout: Timeout = Timeout(2 second)
implicit val executionContext: ExecutionContext = system.dispatcher
val unconfirmed = mutable.Set.empty[Long]
val connFactory = new ConnectionFactory()
connFactory.setHost("127.0.0.1")
connFactory.setPort(5672)
connFactory.setUsername("amqp")
connFactory.setPassword("pass")
val pubConnActor = system.actorOf(ConnectionActor.props(connFactory))
Thread.sleep(5000)
@tailrec
def produce(chActor: ActorRef): Unit = {
chActor ! ChannelMessage({
ch =>
println("------------------------")
ch.basicPublish("amq.direct", "me-topic", MessageProperties.PERSISTENT_BASIC, "message".getBytes("UTF-8"))
}, dropIfNoChannel = false)
produce(chActor)
}
(pubConnActor ? CreateChannel(ChannelActor.props().withMailbox("bounded-mailbox"))).mapTo[ChannelCreated] map {
case ChannelCreated(chActor) =>
produce(chActor)
}
}
When I run my code and after a duration of time stop rabbitmq and after a duration of time start again I get heap space error and stop my program.
I think this is for loop function in ChannelActor.
Please fix it and choose another way.
The text was updated successfully, but these errors were encountered:
If rabbitmq goes offline for a short moment of time (small network glitch for instance), the fallback mechanism is to queue messages in memory for as long as it is offline. But there are limits to the number of messages that can be queued. Is this why you are using a bounded-mailbox?
Note: I am not the maintainer of this library, but more curious about the behaviour you are describing
Thank you.
I just use bounded-mailbox for control message rate.
When I send many messages to channel actor if I don't use a bounded-mailbox, channel actor crash.
A version 5.0.4-beta has been released. While the changes aren't directly related to your problem, it might change the behaviour you are seeing. But as mentioned before, akka-rabbitmq can only queue messages for a limited period of time (that is until the heap space runs out). If the connection to the server is restored within that time, then you should be fine.
Hi
When I send many ChannelMessage with a high rate to ChannelActor with
dropIfNoChannel = false
if I stop rabbitmq and after a duration of time like 1 minute start rabbitmq again, the ChannelActor blocked for process many messages and can't process other message and throw heap space error.I write this code:
When I run my code and after a duration of time stop rabbitmq and after a duration of time start again I get heap space error and stop my program.
I think this is for loop function in ChannelActor.
Please fix it and choose another way.
The text was updated successfully, but these errors were encountered: