Skip to content

Refactor AbstractMessageStream and implementations#4396

Open
hjohn wants to merge 2 commits intoaxon-5.1.xfrom
feature/abstract-deluxe-message-stream
Open

Refactor AbstractMessageStream and implementations#4396
hjohn wants to merge 2 commits intoaxon-5.1.xfrom
feature/abstract-deluxe-message-stream

Conversation

@hjohn
Copy link
Copy Markdown
Contributor

@hjohn hjohn commented Apr 8, 2026 •
edited
Loading

  • Move most of the complicated logic to AbstractMessageStream for streams deriving from it
  • Made AbstractMessageStream safe to extend without breaking its invariants
  • Based FluxMessageStream on AbstractMessageStream
  • Based ConcatenatingMessageStream on AbstractMessageStream
  • Simplified ConcatenatingMessageStream using active stream logic
  • Removed unused code from QueueMessageStream and renamed methods for clarity

By doing the above, this PR resolves #4356

This also fixes a bug in DefaultEventStoreTransaction that I ran into while fixing streams:

Resolves #4200

@hjohn hjohn requested a review from a team as a code owner April 8, 2026 15:59
@hjohn hjohn requested review from MateuszNaKodach, abuijze and corradom and removed request for a team April 8, 2026 15:59
@hjohn hjohn mentioned this pull request Apr 8, 2026
Closed
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch from 4415738 to 4809dc0 Compare April 8, 2026 16:01
@hjohn hjohn self-assigned this Apr 8, 2026
@hjohn hjohn added the Priority 2: Should High priority. Ideally, these issues are part of the release they’re assigned to. label Apr 8, 2026
@smcvb smcvb added this to the Release 5.2.0 milestone Apr 9, 2026
@smcvb smcvb added Type: Bug Use to signal issues that describe a bug within the system. Type: Enhancement Use to signal an issue enhances an already existing feature of the project. labels Apr 9, 2026
smcvb
smcvb requested changes Apr 9, 2026
View reviewed changes
Copy link
Copy Markdown
Contributor

@smcvb smcvb left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

First and foremost: tremendous simplification of the concrete stream implementations! On the other end, the AbstractMessageStream became a tad more complex, let alone concretely validating the shift in a PR. I do have a bunch of concerns and some questions. Honestly not 100% sure if they all merit changes, but I am going to be protective here given the role of the MessageStream.

Comment thread messaging/src/main/java/org/axonframework/messaging/core/AbstractMessageStream.java Outdated
Comment thread messaging/src/main/java/org/axonframework/messaging/core/AbstractMessageStream.java Outdated
Comment thread messaging/src/main/java/org/axonframework/messaging/core/AbstractMessageStream.java Outdated
Comment thread messaging/src/main/java/org/axonframework/messaging/core/AbstractMessageStream.java Outdated
Comment thread messaging/src/main/java/org/axonframework/messaging/core/AbstractMessageStream.java
Comment thread messaging/src/test/java/org/axonframework/messaging/core/ConcatenatingMessageStreamTest.java
Comment thread messaging/src/test/java/org/axonframework/messaging/core/SingleValueMessageStreamTest.java Outdated
Comment thread messaging/src/test/java/org/axonframework/messaging/core/DelayedMessageStreamTest.java
Comment thread messaging/src/test/java/org/axonframework/messaging/core/DelayedMessageStreamTest.java
Comment thread messaging/src/test/java/org/axonframework/messaging/core/ConcatenatingMessageStreamTest.java
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch from 4809dc0 to 467d49c Compare April 9, 2026 14:34
@hjohn hjohn changed the base branch from main to axon-5.1.x April 9, 2026 14:34
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch 3 times, most recently from 7b9184f to a9faf4e Compare April 13, 2026 08:58
smcvb
smcvb reviewed Apr 13, 2026
View reviewed changes
Copy link
Copy Markdown
Contributor

@smcvb smcvb left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

I think the use of synchronized could be documented, as it's a well thought off decision. Furthermore, the before-each assertion seems off to me. Lastly, the flux-pre-subscribe is AFAIK not the typical approach of Project Reactor-like components. Hence, I think it's best if we stick to what users expect when using this.

@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch from a9faf4e to 307cb20 Compare April 13, 2026 12:43
smcvb
smcvb approved these changes Apr 13, 2026
View reviewed changes
Copy link
Copy Markdown
Contributor

@smcvb smcvb left a comment

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

My concerns have been addressed, hence I'm approving this pull request.

@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch 6 times, most recently from 68fa755 to 92ccac0 Compare April 15, 2026 11:13
@hjohn hjohn requested a review from smcvb April 15, 2026 11:17
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch 2 times, most recently from a4b160f to f405fd2 Compare April 15, 2026 13:05
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch 4 times, most recently from a229e81 to 89199bf Compare April 16, 2026 02:34
@smcvb smcvb modified the milestones: Release 5.2.0, Release 5.1.0 Apr 16, 2026
@hjohn
Refactor AbstractMessageStream and implementations
97eb7d3 
- Move most of the complicated logic to AbstractMessageStream for
streams deriving from it
- Made AbstractMessageStream safe to extend without breaking its
invariants
- Based FluxMessageStream on AbstractMessageStream
- Based ConcatenatingMessageStream on AbstractMessageStream
- Simplified ConcatenatingMessageStream using active stream logic
- Removed unused code from QueueMessageStream and renamed methods for
clarity
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch from 89199bf to 97eb7d3 Compare April 16, 2026 08:26
@hjohn hjohn force-pushed the feature/abstract-deluxe-message-stream branch from 8f1d6f6 to 4923a30 Compare April 16, 2026 13:14
@hjohn hjohn linked an issue Apr 16, 2026 that may be closed by this pull request
DefaultEventStoreTransaction can under some circumstances return a sourced stream without filtering the consistency marker #4200
Open
MateuszNaKodach
MateuszNaKodach reviewed Apr 16, 2026
View reviewed changes
Comment thread messaging/src/main/java/org/axonframework/messaging/core/SingleValueMessageStream.java
invokeCallbackSafely();
}
});
this.source.thenAccept(e -> signalProgress());
Copy link
Copy Markdown
Contributor

@MateuszNaKodach MateuszNaKodach Apr 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bug: missing progress signal on exceptional future completion

Suggested change
this.source.thenAccept(e -> signalProgress());
this.source.whenComplete((entry, throwable) -> signalProgress());

thenAccept only fires on successful completion. When the CompletableFuture completes exceptionally, signalProgress() is never called. If a consumer has registered a callback (via setCallback or reduce()), awaitingData = true, and
the callback will never fire — causing the stream to hang permanently.

fetchNext() already handles isCompletedExceptionally() correctly, so just the signal is missing.

Comment thread messaging/src/main/java/org/axonframework/messaging/core/FluxMessageStream.java
@Override
public void onError(Throwable t) {
peeked.add(FetchResult.error(t));
seal();
Copy link
Copy Markdown
Contributor

@MateuszNaKodach MateuszNaKodach Apr 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

bug: missing signalProgress() call

in onComplete we have signalProgress() at the end of the method.
Should we have it here as well?

MateuszNaKodach
MateuszNaKodach reviewed Apr 16, 2026
View reviewed changes
Comment thread ...g/src/main/java/org/axonframework/eventsourcing/eventstore/DefaultEventStoreTransaction.java
Comment on lines +114 to +116
if (!Boolean.TRUE.equals(processingContext.getResource(prepareCommitExecuted))) {
updateAppendPosition(marker);
}
Copy link
Copy Markdown
Contributor

@MateuszNaKodach MateuszNaKodach Apr 16, 2026

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Is it correct description of this change?

The Bug Fix in DefaultEventStoreTransaction

The source() method attaches a handler that updates the append position when a stream completes. But if source() was called multiple times (for multiple sourcing conditions), the onComplete callback fired multiple times, updating
the position each time — causing incorrect behavior.

Fix: a prepareCommitExecuted flag on the ProcessingContext:

Before:
source() call 1 ──► onComplete ──► updateAppendPosition() ✓
source() call 2 ──► onComplete ──► updateAppendPosition() ✗ (ran again!)

After:
source() call 1 ──► onComplete ──► prepareCommitExecuted? NO → updateAppendPosition()
set prepareCommitExecuted = true
source() call 2 ──► onComplete ──► prepareCommitExecuted? YES → skip ✓

Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Reviewers

@MateuszNaKodach MateuszNaKodach MateuszNaKodach left review comments

@abuijze abuijze Awaiting requested review from abuijze abuijze was automatically assigned from AxonFramework/framework-developers

@corradom corradom Awaiting requested review from corradom corradom was automatically assigned from AxonFramework/framework-developers

@smcvb smcvb Awaiting requested review from smcvb

Assignees

@hjohn hjohn

Labels

Priority 2: Should High priority. Ideally, these issues are part of the release they’re assigned to. Type: Bug Use to signal issues that describe a bug within the system. Type: Enhancement Use to signal an issue enhances an already existing feature of the project.

Projects

None yet

Milestone

Release 5.1.0

3 participants

@hjohn @smcvb @MateuszNaKodach