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

  • 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 exclusively by the enclosing AbstractMessageStream implementation. Calls are serialized under a single internal lock; the method is never executed concurrently with itself or other lifecycle methods of this instance. Implementations may assume single-threaded access and do not need to provide their own synchronization for correctness of this method.

      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:

      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.

      This method is invoked exclusively by the enclosing AbstractMessageStream implementation. Calls are serialized under a single internal lock; the method is never executed concurrently with itself or other lifecycle methods of this instance. Implementations may assume single-threaded access and do not need to provide their own synchronization for correctness of this method.

      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