Package io.axoniq.framework.axonserver.connector.query

Class AbstractQueryResponseMessageStream<T>

java.lang.Object

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

io.axoniq.framework.axonserver.connector.query.AbstractQueryResponseMessageStream<T>

Type Parameters:

T - The type of the objects in the underlying ResultStream to be transformed into QueryResponseMessages.

All Implemented Interfaces:

MessageStream<QueryResponseMessage>

Direct Known Subclasses:

QueryResponseMessageStream, QueryUpdateMessageStream

@Internal
public abstract class AbstractQueryResponseMessageStream<T>
extends AbstractMessageStream<QueryResponseMessage>

An abstract implementation of the MessageStream interface that wraps a ResultStream. This class provides functionality for transforming the data in the ResultStream into QueryResponseMessages, handling any encountered errors, and managing stream lifecycle events.

Since:

5.0.0

Author:

Allard Buijze, Jan Gallinkski, 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

Modifier	Constructor	Description
protected	AbstractQueryResponseMessageStream(io.axoniq.axonserver.connector.ResultStream<T> stream)	Constructs an instance of the AbstractQueryResponseMessageStream class with the provided result stream.

Method Summary

Modifier and Type	Method	Description
protected AbstractMessageStream.FetchResult<MessageStream.Entry<QueryResponseMessage>>	fetchNext()	Attempts to fetch the next available MessageStream.Entry from the underlying source.
protected abstract boolean	isError(T t)
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

AbstractQueryResponseMessageStream

protected AbstractQueryResponseMessageStream(io.axoniq.axonserver.connector.ResultStream<T> stream)

Constructs an instance of the AbstractQueryResponseMessageStream class with the provided result stream.

Parameters:

stream - The ResultStream instance from which query response data will be fetched. Must not be null.

Method Details

fetchNext

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

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

isError

protected abstract boolean isError(T t)