object ProducerController
Point-to-point reliable delivery between a single producer actor sending messages and a single consumer actor receiving the messages. Used together with ConsumerController.
The producer actor will start the flow by sending a ProducerController.Start message to
the ProducerController
. The ActorRef
in the Start
message is typically constructed
as a message adapter to map the ProducerController.RequestNext to the protocol of the
producer actor.
For the ProducerController
to know where to send the messages it must be connected with the
ConsumerController
. You do this is with ProducerController.RegisterConsumer or
ConsumerController.RegisterToProducerController messages.
The ProducerController
sends RequestNext
to the producer, which is then allowed to send one
message to the ProducerController
via the sendNextTo
in the RequestNext
. Thereafter the
producer will receive a new RequestNext
when it's allowed to send one more message.
The producer and ProducerController
actors are supposed to be local so that these messages are
fast and not lost. This is enforced by a runtime check.
Many unconfirmed messages can be in flight between the ProducerController
and ConsumerController
.
The flow control is driven by the consumer side, which means that the ProducerController
will
not send faster than the demand requested by the ConsumerController
.
Lost messages are detected, resent and deduplicated if needed. This is also driven by the consumer side,
which means that the ProducerController
will not push resends unless requested by the
ConsumerController
.
Until sent messages have been confirmed the ProducerController
keeps them in memory to be able to
resend them. If the JVM of the ProducerController
crashes those unconfirmed messages are lost.
To make sure the messages can be delivered also in that scenario the ProducerController
can be
used with a DurableProducerQueue. Then the unconfirmed messages are stored in a durable way so
that they can be redelivered when the producer is started again. An implementation of the
DurableProducerQueue
is provided by EventSourcedProducerQueue
in akka-persistence-typed
.
Instead of using tell
with the sendNextTo
in the RequestNext
the producer can use context.ask
with the askNextTo
in the RequestNext
. The difference is that a reply is sent back when the
message has been handled. If a DurableProducerQueue
is used then the reply is sent when the message
has been stored successfully, but it might not have been processed by the consumer yet. Otherwise the
reply is sent after the consumer has processed and confirmed the message.
If the consumer crashes a new ConsumerController
can be connected to the original ProducerConsumer
without restarting it. The ProducerConsumer
will then redeliver all unconfirmed messages.
It's also possible to use the ProducerController
and ConsumerController
without resending
lost messages, but the flow control is still used. This can for example be useful when both consumer and
producer are know to be located in the same local ActorSystem
. This can be more efficient since messages
don't have to be kept in memory in the ProducerController
until they have been
confirmed, but the drawback is that lost messages will not be delivered. See configuration
only-flow-control
of the ConsumerController
.
The producerId
is used in logging and included as MDC entry with key "producerId"
. It's propagated
to the ConsumerController
and is useful for correlating log messages. It can be any String
but it's
recommended to use a unique identifier of representing the producer.
If the DurableProducerQueue
is defined it is created as a child actor of the ProducerController
actor.
It will use the same dispatcher as the parent ProducerController
.
- Annotations
- @ApiMayChange()
- Source
- ProducerController.scala
- Alphabetic
- By Inheritance
- ProducerController
- AnyRef
- Any
- Hide All
- Show All
- Public
- Protected
Type Members
- sealed trait Command[A] extends UnsealedInternalCommand
- final case class MessageWithConfirmation[A](message: A, replyTo: ActorRef[SeqNr]) extends UnsealedInternalCommand with Product with Serializable
For sending confirmation message back to the producer when the message has been confirmed.
For sending confirmation message back to the producer when the message has been confirmed. Typically used with
context.ask
from the producer.If
DurableProducerQueue
is used the confirmation reply is sent when the message has been successfully stored, meaning that the actual delivery to the consumer may happen later. IfDurableProducerQueue
is not used the confirmation reply is sent when the message has been fully delivered, processed, and confirmed by the consumer. - final case class RegisterConsumer[A](consumerController: ActorRef[ConsumerController.Command[A]]) extends Command[A] with DeliverySerializable with Product with Serializable
Register the given
consumerController
to theProducerController
.Register the given
consumerController
to theProducerController
.Alternatively, this registration can be done on the consumer side with the ConsumerController.RegisterToProducerController message.
When using a custom
send
function for theProducerController
this should not be used. - final case class RequestNext[A](producerId: String, currentSeqNr: SeqNr, confirmedSeqNr: SeqNr, sendNextTo: ActorRef[A], askNextTo: ActorRef[MessageWithConfirmation[A]]) extends Product with Serializable
The
ProducerController
sendsRequestNext
to the producer when it is allowed to send one message via thesendNextTo
oraskNextTo
.The
ProducerController
sendsRequestNext
to the producer when it is allowed to send one message via thesendNextTo
oraskNextTo
. Note that only one message is allowed, and then it must wait for nextRequestNext
before sending one more message. - type SeqNr = Long
- final class Settings extends AnyRef
- final case class Start[A](producer: ActorRef[RequestNext[A]]) extends Command[A] with Product with Serializable
Initial message from the producer actor.
Initial message from the producer actor. The
producer
is typically constructed as a message adapter to map the RequestNext to the protocol of the producer actor.If the producer is restarted it should send a new
Start
message to theProducerController
.
Value Members
- final def !=(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- final def ##: Int
- Definition Classes
- AnyRef → Any
- final def ==(arg0: Any): Boolean
- Definition Classes
- AnyRef → Any
- def apply[A](producerId: String, durableQueueBehavior: Option[Behavior[DurableProducerQueue.Command[A]]], settings: Settings)(implicit arg0: ClassTag[A]): Behavior[Command[A]]
- def apply[A](producerId: String, durableQueueBehavior: Option[Behavior[DurableProducerQueue.Command[A]]])(implicit arg0: ClassTag[A]): Behavior[Command[A]]
- final def asInstanceOf[T0]: T0
- Definition Classes
- Any
- def clone(): AnyRef
- Attributes
- protected[lang]
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.CloneNotSupportedException]) @HotSpotIntrinsicCandidate() @native()
- def create[A](messageClass: Class[A], producerId: String, durableQueueBehavior: Optional[Behavior[DurableProducerQueue.Command[A]]], settings: Settings): Behavior[Command[A]]
Java API
- def create[A](messageClass: Class[A], producerId: String, durableQueueBehavior: Optional[Behavior[DurableProducerQueue.Command[A]]]): Behavior[Command[A]]
Java API
- final def eq(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- def equals(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef → Any
- final def getClass(): Class[_ <: AnyRef]
- Definition Classes
- AnyRef → Any
- Annotations
- @HotSpotIntrinsicCandidate() @native()
- def hashCode(): Int
- Definition Classes
- AnyRef → Any
- Annotations
- @HotSpotIntrinsicCandidate() @native()
- final def isInstanceOf[T0]: Boolean
- Definition Classes
- Any
- final def ne(arg0: AnyRef): Boolean
- Definition Classes
- AnyRef
- final def notify(): Unit
- Definition Classes
- AnyRef
- Annotations
- @HotSpotIntrinsicCandidate() @native()
- final def notifyAll(): Unit
- Definition Classes
- AnyRef
- Annotations
- @HotSpotIntrinsicCandidate() @native()
- def requestNextClass[A](): Class[RequestNext[A]]
Java API: The generic
Class
type forProducerController.RequestNext
that can be used when creating amessageAdapter
forClass<RequestNext<MessageType>>
. - final def synchronized[T0](arg0: => T0): T0
- Definition Classes
- AnyRef
- def toString(): String
- Definition Classes
- AnyRef → Any
- final def wait(arg0: Long, arg1: Int): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- final def wait(arg0: Long): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException]) @native()
- final def wait(): Unit
- Definition Classes
- AnyRef
- Annotations
- @throws(classOf[java.lang.InterruptedException])
- object Settings