Package org.axonframework.messaging.core

Class CloseCallbackMessageStream<M extends Message>

java.lang.Object

org.axonframework.messaging.core.AbstractMessageStream<M>

org.axonframework.messaging.core.CloseCallbackMessageStream<M>

Type Parameters:

M - The type of Message handled by this MessageStream.

All Implemented Interfaces:

MessageStream<M>

public class CloseCallbackMessageStream<M extends Message>
extends AbstractMessageStream<M>

MessageStream implementation that invokes the given closeHandler once the stream is closed. A stream is considered closed when a consumer explicitly calls AbstractMessageStream.close() or when the stream is completed.

Note that when close is called on the delegate, or when the client does not attempt to consume this stream, the close handler may never be invoked, even though the stream is completed.

Since:

5.0.0

Author:

John Hendrikx

Nested Class Summary

Nested classes/interfaces inherited from class org.axonframework.messaging.core.AbstractMessageStream

AbstractMessageStream.FetchResult<T extends @Nullable MessageStream.Entry<?>>

Nested classes/interfaces inherited from interface org.axonframework.messaging.core.MessageStream

MessageStream.Empty<M extends Message>, MessageStream.Entry<M extends Message>, MessageStream.Single<M extends Message>

Constructor Summary

Constructors

Constructor	Description
CloseCallbackMessageStream(MessageStream<M> delegate, Runnable closeHandler)	Creates an instance of the CloseCallbackMessageStream, calling the given closeHandler once this stream is closed, or the given delegate completes.

Method Summary

Modifier and Type	Method	Description
String	describeDelegates()	Subtypes should override this to describe any (message stream) delegates they use for debugging purposes.
protected AbstractMessageStream.FetchResult<MessageStream.Entry<M>>	fetchNext()	Attempts to fetch the next available MessageStream.Entry from the underlying source.
protected void	onCompleted()	Callback invoked when the stream is about to transition to a completed state, either successfully or exceptionally.

Methods inherited from class org.axonframework.messaging.core.AbstractMessageStream

close, describeFlags, error, hasNextAvailable, initialize, isCompleted, next, peek, setCallback, signalProgress, toString

Methods inherited from class java.lang.Object

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

Methods inherited from interface org.axonframework.messaging.core.MessageStream

cast, concatWith, filter, first, ignoreEntries, map, mapMessage, onClose, onComplete, onErrorContinue, onNext, reduce

Constructor Details

CloseCallbackMessageStream

public CloseCallbackMessageStream(MessageStream<M> delegate,
 Runnable closeHandler)

Creates an instance of the CloseCallbackMessageStream, calling the given closeHandler once this stream is closed, or the given delegate completes.

Parameters:

delegate - The MessageStream to wrap with the close handler invocation logic

closeHandler - The handler to invoke when the stream is closed or completed

Method Details

fetchNext

protected AbstractMessageStream.FetchResult<MessageStream.Entry<M>> fetchNext()

Description copied from class: AbstractMessageStream

Attempts to fetch the next available MessageStream.Entry from the underlying source.

This method is invoked by AbstractMessageStream.next() when no previously peeked entry is available. Implementations must return a AbstractMessageStream.FetchResult describing the current state of the stream:

• AbstractMessageStream.FetchResult.Value if an entry is immediately available,
• AbstractMessageStream.FetchResult.NotReady if no entry is currently available but more may arrive later,
• AbstractMessageStream.FetchResult.Completed if the stream is exhausted and will produce no further entries,
• AbstractMessageStream.FetchResult.Error if the stream has failed with an error.

Returning AbstractMessageStream.FetchResult.NotReady will transition the stream into an awaiting data state. Implementations must subsequently invoke AbstractMessageStream.signalProgress() when progress may be possible again (e.g., when new data arrives or the stream completes).

Implementations must ensure that any state changes observable via this method are fully applied before invoking AbstractMessageStream.signalProgress(). A signal that arrives before the consumer has entered the awaiting state is not replayed; correctness relies on this method returning the updated state when the consumer calls it next. See AbstractMessageStream.signalProgress() for the full ordering contract.

This method must be non-blocking. It should return immediately with the best available information about the stream's current state.

Implementations must not attempt to complete or close the stream directly. Instead, they must return AbstractMessageStream.FetchResult.Completed or AbstractMessageStream.FetchResult.Error to signal termination.

If an implementation throws an exception, the stream completes exceptionally with that exception.

Specified by:

fetchNext in class AbstractMessageStream<M extends Message>

Returns:

a AbstractMessageStream.FetchResult representing the outcome of the fetch attempt

onCompleted

protected void onCompleted()

Description copied from class: AbstractMessageStream

Callback invoked when the stream is about to transition to a completed state, either successfully or exceptionally. Subclasses may override this method to perform custom actions on completion.

If the implementation throws an exception, the stream still completes, but it will complete with the thrown exception. If the stream was about to complete with an error, and the callback fails as well, the exception is added as a suppressed exception.

Overrides:

onCompleted in class AbstractMessageStream<M extends Message>

describeDelegates

public String describeDelegates()

Description copied from class: AbstractMessageStream

Subtypes should override this to describe any (message stream) delegates they use for debugging purposes. This allows to visualize a chain of message streams, their states and how they are linked.

If there are multiple delegates, then they should be comma separated with the active delegate prepended with an asterisk ("*").

Overrides:

describeDelegates in class AbstractMessageStream<M extends Message>

Returns:

the description of delegate streams, or null if there are no delegates