Package org.axonframework.messaging.core

Class DelegatingMessageStream<DM extends Message,RM extends Message>

java.lang.Object
org.axonframework.messaging.core.DelegatingMessageStream<DM,RM>
Type Parameters:
DM - The type of Message handled by the delegate.
RM - The type of Message handled by this MessageStream.
All Implemented Interfaces:
MessageStream<RM>
public abstract class DelegatingMessageStream<DM extends Message,RM extends Message> extends Object implements MessageStream<RM>
Abstract implementation of an MessageStream that delegates calls to a given delegate.
Since:
5.0.0
Author:
Allard Buijze

  • Constructor Details

    • DelegatingMessageStream

      public DelegatingMessageStream(MessageStream<DM> delegate)
      Constructs the DelegatingMessageStream with given delegate to receive calls.
      Parameters:
      delegate - The instance to delegate calls to.

  • Method Details

    • setCallback

      public void setCallback(Runnable callback)
      Description copied from interface: MessageStream
      Registers the callback to invoke when entries are available for reading or when the stream completes (either normally or with an error), with the intent that a consumer of this stream will read the available messages completely.

      This has the following implications:

      • When you register the callback on an already completed stream, the callback is invoked directly.
      • When you register the callback on a stream having entries available, the callback is invoked, and you must consume the existing entries.
      • A registered callback is invoked again when all entries were consumed and new entries arrive.
      • Depending on the implementation of the stream, your callback might be invoked again while you are still consuming entries, in these cases it is your responsibility to synchronize the callback executions. Using MessageStream.reduce(Object, BiFunction) might be a better fit because it guarantees isolated execution of callbacks using a processingGate.

      Note that an invocation of the callback does not in any way guarantee that entries are indeed available, or that the stream has indeed been completed. Implementations may choose to suppress repeated invocations of the callback if no entries have been read in the meantime.

      Any previously registered callback is replaced with the given callback.

      The callback is called on an arbitrary thread, and it should keep work performed on this thread to a minimum as this may interfere with other callbacks handled by the same thread. Any exception thrown by the callback will result in the stream completing with this exception as the error, unless the callback was called to indicate completion.

      Specified by:
      setCallback in interface MessageStream<DM extends Message>
      Parameters:
      callback - The callback to invoke when entries are available for reading, or the stream completes.
      See Also:

    • error

      public Optional<Throwable> error()
      Description copied from interface: MessageStream
      Indicates whether any error has been reported in this stream. Implementations may choose to not return any error here until all entries that were available for reading before any error occurred have been consumed.
      Specified by:
      error in interface MessageStream<DM extends Message>
      Returns:
      An optional containing the possible error this stream completed with.

    • isCompleted

      public boolean isCompleted()
      Description copied from interface: MessageStream
      Indicates whether this stream has been completed. A completed stream will never return entries from MessageStream.next(), and MessageStream.hasNextAvailable() will always return false. If the stream completed with an error, MessageStream.error() will report so.
      Specified by:
      isCompleted in interface MessageStream<DM extends Message>
      Returns:
      true if the stream completed, otherwise false.

    • hasNextAvailable

      public boolean hasNextAvailable()
      Description copied from interface: MessageStream
      Indicates whether an entry is available for immediate reading. When entries are reported available, there is no guarantee that MessageStream.next() will indeed return an entry. However, besides any concurrent activity on this stream, it is guaranteed that no entries are available for reading when this method returns false.
      Specified by:
      hasNextAvailable in interface MessageStream<DM extends Message>
      Returns:
      true when there are entries available for reading, false otherwise.

    • close

      public void close()
      Description copied from interface: MessageStream
      Closes this stream, indicating that the consumer is no longer interested in receiving further entries.

      This is a consumer-driven cancellation operation. It immediately transitions the stream into a terminal state and discards any remaining or buffered entries, including any peeked entries. After this call returns:

      This method is intended exclusively for consumer-side cancellation. It signals that no further processing is required and allows implementations to release all resources immediately.

      This operation is fundamentally different from producer-side completion. When an implementation determines that no further elements will ever become available, it must not call close(). Instead, it must:

      • stop producing new elements,
      • allow any buffered elements to be consumed normally, and
      • transition the stream to completion using its normal terminal signaling mechanism (e.g. completion or error through next()/fetchNext()).

      Calling this method is only required when the consumer chooses not to fully consume the stream. Streams must always release resources when they complete, regardless of whether close() is invoked.

      Specified by:
      close in interface MessageStream<DM extends Message>

    • delegate

      protected MessageStream<DM> delegate()
      Returns the delegate as provided in the constructor.
      Returns:
      the delegate as provided in the constructor.