Package org.axonframework.eventsourcing.eventstore

Class ContinuousMessageStream<E>

java.lang.Object

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

org.axonframework.eventsourcing.eventstore.ContinuousMessageStream<E>

Type Parameters:

E - the type of the raw elements returned by the fetcher before conversion to EventMessages

All Implemented Interfaces:

MessageStream<EventMessage>

@Internal
public final class ContinuousMessageStream<E>
extends AbstractMessageStream<EventMessage>

A MessageStream implementation that continuously fetches event messages from a configurable data source. This stream has no defined end and will continue retrieving new batches of data as they become available.

The stream relies on externally provided functional strategies to control its behavior:

• A fetcher to obtain the next batch of elements. The fetcher owns all cursor/position state internally and is simply called each time more data is needed.
• A converter to transform fetched elements into MessageStream.Entry instances.
• A callbackTracker to manage callback registration for new data availability.

The stream supports lifecycle management through AbstractMessageStream.close() and callback registration via AbstractMessageStream.setCallback(Runnable).

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
ContinuousMessageStream(Supplier<List<E>> fetcher, Function<E,MessageStream.Entry<EventMessage>> converter, BiFunction<ContinuousMessageStream<?>,Runnable,Registration> callbackTracker)	Creates a new ContinuousMessageStream instance configured with the given strategies.

Method Summary

Modifier and Type	Method	Description
protected AbstractMessageStream.FetchResult<MessageStream.Entry<EventMessage>>	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, describeDelegates, 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

ContinuousMessageStream

public ContinuousMessageStream(Supplier<List<E>> fetcher,
 Function<E,MessageStream.Entry<EventMessage>> converter,
 BiFunction<ContinuousMessageStream<?>,Runnable,Registration> callbackTracker)

Creates a new ContinuousMessageStream instance configured with the given strategies.

Parameters:

fetcher - a supplier that returns the next batch of elements to emit. The fetcher is responsible for tracking its own position; it is called repeatedly whenever the stream needs more data. Must not return null, but may return an empty list to indicate no new data is currently available.

converter - a function converting each fetched element into an MessageStream.Entry containing an EventMessage

callbackTracker - a function that, given this stream and a callback Runnable, registers a listener and returns a Registration allowing it to be canceled

Method Details

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<EventMessage>

fetchNext

protected AbstractMessageStream.FetchResult<MessageStream.Entry<EventMessage>> 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<EventMessage>

Returns:

a AbstractMessageStream.FetchResult representing the outcome of the fetch attempt