Package org.apache.pekko.actor.typed.delivery.internal

Class ConsumerControllerImpl$

java.lang.Object

org.apache.pekko.actor.typed.delivery.internal.ConsumerControllerImpl$

public class ConsumerControllerImpl$
extends Object

INTERNAL API

==== Design notes ====

The destination consumer will start the flow by sending an initial Start message to the ConsumerController.

The ProducerController sends the first message to the ConsumerController without waiting for a Request from the ConsumerController. The main reason for this is that when used with Cluster Sharding the first message will typically create the ConsumerController. It's also a way to connect the ProducerController and ConsumerController in a dynamic way, for example when the ProducerController is replaced.

The ConsumerController sends ProducerControllerImpl.Request to the ProducerController to specify it's ready to receive up to the requested sequence number.

The ConsumerController sends the first Request when it receives the first SequencedMessage and has received the Start message from the consumer.

It sends new Request when half of the requested window is remaining, but it also retries the Request if no messages are received because that could be caused by lost messages.

Apart from the first message the producer will not send more messages than requested.

Received messages are wrapped in ConsumerController.Delivery when sent to the consumer, which is supposed to reply with ConsumerController.Confirmed when it has processed the message. Next message is not delivered until the previous is confirmed. More messages from the producer that arrive while waiting for the confirmation are stashed by the ConsumerController and delivered when previous message was confirmed.

In other words, the "request" protocol to the application producer and consumer is one-by-one, but between the ProducerController and ConsumerController it's window of messages in flight.

The consumer and the ConsumerController are supposed to be local so that these messages are fast and not lost.

If the ConsumerController receives a message with unexpected sequence number (not previous + 1) it sends ProducerControllerImpl.Resend to the ProducerController and will ignore all messages until the expected sequence number arrives.

Field Summary

Fields

Modifier and Type	Field	Description
static final ConsumerControllerImpl$	MODULE$	Static reference to the singleton instance of this Scala object.

Constructor Summary

Constructors

Constructor	Description
ConsumerControllerImpl$()

Method Summary

Modifier and Type	Method	Description
<A> Behavior<ConsumerController.Command<A>>	apply(scala.Option<ServiceKey<ConsumerController.Command<A>>> serviceKey, ConsumerController.Settings settings)
void	enforceLocalConsumer(ActorRef<?> ref)

Methods inherited from class java.lang.Object

clone, equals, finalize, getClass, hashCode, notify, notifyAll, toString, wait, wait, wait

Field Details

MODULE$

public static final ConsumerControllerImpl$ MODULE$

Static reference to the singleton instance of this Scala object.

Constructor Details

ConsumerControllerImpl$

public ConsumerControllerImpl$()

Method Details

apply

public <A> Behavior<ConsumerController.Command<A>> apply(scala.Option<ServiceKey<ConsumerController.Command<A>>> serviceKey,
 ConsumerController.Settings settings)

enforceLocalConsumer

public void enforceLocalConsumer(ActorRef<?> ref)