Class ActorPublisher<T>

Extend this actor to make it a stream publisher that keeps track of the subscription life cycle and requested elements.

Create a Reactive.Streams.IPublisher<T> backed by this actor with Create<T>(IActorRef).

It can be attached to a Reactive.Streams.ISubscriber<T> or be used as an input source for a IFlow<TOut, TMat>. You can only attach one subscriber to this publisher.

The life cycle state of the subscription is tracked with the following boolean members: IsActive, IsCompleted, IsErrorEmitted, and IsCanceled.

You send elements to the stream by calling OnNext(T). You are allowed to send as many elements as have been requested by the stream subscriber. This amount can be inquired with TotalDemand. It is only allowed to use OnNext(T) when IsActive TotalDemand > 0, otherwise OnNext(T) will throw IllegalStateException.

When the stream subscriber requests more elements the Request message is delivered to this actor, and you can act on that event. The TotalDemand is updated automatically.

When the stream subscriber cancels the subscription the Cancel message is delivered to this actor. After that subsequent calls to OnNext(T) will be ignored.

You can complete the stream by calling OnComplete(). After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

You can terminate the stream with failure by calling OnError(Exception). After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

If you suspect that this ActorPublisher<T> may never get subscribed to, you can override the SubscriptionTimeout method to provide a timeout after which this Publisher should be considered canceled. The actor will be notified when the timeout triggers via an SubscriptionTimeoutExceeded message and MUST then perform cleanup and stop itself.

If the actor is stopped the stream will be completed, unless it was not already terminated with failure, completed or canceled.

Inheritance

object

ActorBase

ActorPublisher<T>

Implements

IInternalActor

Inherited Members

ActorBase.Sender

ActorBase.Self

ActorBase.Context

ActorBase.Receive(object)

ActorBase.EmptyReceive

ActorBase.Unhandled(object)

ActorBase.Become(Receive)

ActorBase.BecomeStacked(Receive)

ActorBase.UnbecomeStacked()

ActorBase.SetReceiveTimeout(TimeSpan?)

ActorBase.PreStart()

ActorBase.PreRestart(Exception, object)

ActorBase.PostRestart(Exception)

ActorBase.PostStop()

ActorBase.SupervisorStrategy()

object.Equals(object)

object.Equals(object, object)

object.GetHashCode()

object.GetType()

object.MemberwiseClone()

object.ReferenceEquals(object, object)

object.ToString()

Namespace: Akka.Streams.Actors

Assembly: Akka.Streams.dll

Syntax

public abstract class ActorPublisher<T> : ActorBase, IInternalActor

Type Parameters

Name	Description
T	TBD

Constructors

ActorPublisher()

TBD

Declaration

protected ActorPublisher()

Properties

IsActive

The state when the publisher is active, i.e. before the subscriber is attached and when an subscriber is attached. It is allowed to call OnComplete() and OnError(Exception) in this state. It is allowed to call OnNext(T) in this state when TotalDemand is greater than zero.

Declaration

public bool IsActive { get; }

Property Value

Type	Description
bool

IsCanceled

The state after the stream subscriber has canceled the subscription. It is allowed to call OnNext(T), OnError(Exception), and OnComplete() in this state, but the calls will not perform anything.

Declaration

public bool IsCanceled { get; }

Property Value

Type	Description
bool

IsCompleted

The terminal state after calling OnComplete(). It is not allowed to OnNext(T), OnError(Exception), and OnComplete() in this state.

Declaration

public bool IsCompleted { get; }

Property Value

Type	Description
bool

IsErrorEmitted

The terminal state after calling OnError(Exception). It is not allowed to call OnNext(T), OnError(Exception), and OnComplete() in this state.

Declaration

public bool IsErrorEmitted { get; }

Property Value

Type	Description
bool

SubscriptionTimeout

Subscription timeout after which this actor will become Canceled and reject any incoming "late" subscriber.

The actor will receive an SubscriptionTimeoutExceeded message upon which it MUST react by performing all necessary cleanup and stopping itself.

Use this feature in order to avoid leaking actors when you suspect that this Publisher may never get subscribed to by some Subscriber.

Declaration

public TimeSpan SubscriptionTimeout { get; protected set; }

Property Value

Type	Description
TimeSpan

TotalDemand

Total number of requested elements from the stream subscriber. This actor automatically keeps tracks of this amount based on incoming request messages and outgoing OnNext(T).

Declaration

public long TotalDemand { get; }

Property Value

Type	Description
long

Methods

AroundPostRestart(Exception, object)

TBD

Declaration

public override void AroundPostRestart(Exception cause, object message)

Parameters

Type	Name	Description
Exception	cause	TBD
object	message	TBD

Overrides

ActorBase.AroundPostRestart(Exception, object)

AroundPostStop()

TBD

Declaration

public override void AroundPostStop()

Overrides

ActorBase.AroundPostStop()

AroundPreRestart(Exception, object)

TBD

Declaration

public override void AroundPreRestart(Exception cause, object message)

Parameters

Type	Name	Description
Exception	cause	TBD
object	message	TBD

Overrides

ActorBase.AroundPreRestart(Exception, object)

AroundPreStart()

TBD

Declaration

public override void AroundPreStart()

Overrides

ActorBase.AroundPreStart()

AroundReceive(Receive, object)

TBD

Declaration

protected override bool AroundReceive(Receive receive, object message)

Parameters

Type	Name	Description
Receive	receive	TBD
object	message	TBD

Returns

Type	Description
bool	TBD

Overrides

ActorBase.AroundReceive(Receive, object)

OnComplete()

Complete the stream. After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

Declaration

public void OnComplete()

Exceptions

Type	Condition
IllegalStateException	This exception is thrown for a number of reasons. These include: when in the ErrorEmitted state This exception is thrown when this ActorPublisher<T> has already terminated due to an error. when in the Completed or CompleteThenStop state This exception is thrown when this ActorPublisher<T> has already completed.

OnCompleteThenStop()

Complete the stream. After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

After signalling completion the Actor will then stop itself as it has completed the protocol. When OnComplete() is called before any Reactive.Streams.ISubscriber<T> has had the chance to subscribe to this ActorPublisher<T> the completion signal (and therefore stopping of the Actor as well) will be delayed until such Reactive.Streams.ISubscriber<T> arrives.

Declaration

public void OnCompleteThenStop()

OnError(Exception)

Terminate the stream with failure. After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

Declaration

public void OnError(Exception cause)

Parameters

Type	Name	Description
Exception	cause	TBD

Exceptions

Type	Condition
IllegalStateException	This exception is thrown for a number of reasons. These include: when in the ErrorEmitted state This exception is thrown when this ActorPublisher<T> has already terminated due to an error. when in the Completed or CompleteThenStop state This exception is thrown when this ActorPublisher<T> has already completed.

OnErrorThenStop(Exception)

Terminate the stream with failure. After that you are not allowed to call OnNext(T), OnError(Exception) and OnComplete().

After signalling the Error the Actor will then stop itself as it has completed the protocol. When OnError(Exception) is called before any Reactive.Streams.ISubscriber<T> has had the chance to subscribe to this ActorPublisher<T> the error signal (and therefore stopping of the Actor as well) will be delayed until such Reactive.Streams.ISubscriber<T> arrives.

Declaration

public void OnErrorThenStop(Exception cause)

Parameters

Type	Name	Description
Exception	cause	TBD

OnNext(T)

Sends an element to the stream subscriber. You are allowed to send as many elements as have been requested by the stream subscriber. This amount can be inquired with TotalDemand. It is only allowed to use OnNext(T) when IsActive and TotalDemand > 0, otherwise OnNext(T) will throw IllegalStateException.

Declaration

public void OnNext(T element)

Parameters

Type	Name	Description
T	element	TBD

Exceptions

Type

Condition

IllegalStateException

This exception is thrown for a number of reasons. These include: when in the Active or PreSubscriber state This exception is thrown when the ActorPublisher<T> has zero TotalDemand. when in the ErrorEmitted state This exception is thrown when this ActorPublisher<T> has already terminated due to an error. when in the Completed or CompleteThenStop state This exception is thrown when this ActorPublisher<T> has already completed.

Implements

IInternalActor

Extension Methods

ObjectExtensions.IsDefaultForType<T>(T)

ObjectExtensions.AsOption<T>(T)

Extensions.AsInstanceOf<T>(object)