Skip to content

4.x: Streamable API: DispatchStreamProcessor & other improvements#8217

Merged
akarnokd merged 3 commits into
ReactiveX:4.xfrom
akarnokd:4xStreamableProcessors1
Jul 6, 2026
Merged

4.x: Streamable API: DispatchStreamProcessor & other improvements#8217
akarnokd merged 3 commits into
ReactiveX:4.xfrom
akarnokd:4xStreamableProcessors1

Conversation

@akarnokd

@akarnokd akarnokd commented Jul 6, 2026

Copy link
Copy Markdown
Member

New types

StreamerInput

Represents an awaitable push side of signaling items and terminal states.

ℹ️ We might rename this to StreamObserver or StreamSubscriber?

StreamProcessor

Combines a Streamable and a StreamerInput to provide a similar interface to Subject and Flow.Processors for the Streamable world.

DispatchStreamProcessor

A publishing / multicasting processor where pushing events have to be awaited until they go through to all current Streamers (or fall through if there are none).

Test API improvements

TestSubscriber / TestObserver + awaitOnSubscribe

Created a latched await for receiving onSubscribe events because in the Streamable world, subscribe and test run on a background thread and thus the whole lifecycle is shifted in time. This call will make sure the upstream reached a point where it called onSubscribe.

Tests updated accordingly.

TestSubscriber / TestObserver + awaitCount(int atLeast, long timeoutMillis, long sleepMillis)

The plain awaitCount(int atLeast) was introduced a while ago with predefined 5000 milliseconds timeout and 10 sleeps. This overload makes the times configurable for faster test execution, especially if the call was supposed to fail.

Bugfixes

  • Streamable.subscribe(Subscriber) did not properly wire up the cancellation mechanisms.

@github-actions

github-actions Bot commented Jul 6, 2026

Copy link
Copy Markdown

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.586 bits/char
Verdict: ✅ Mid-4 beauty detected (thoughtful human code!)

Changed files entropy:

src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java: 4.960
src/main/java/io/reactivex/rxjava4/core/Streamable.java: 4.655
src/main/java/io/reactivex/rxjava4/core/StreamerInput.java: 4.741
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java: 4.407
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDelay.java: 4.580
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java: 4.871
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java: 4.316
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java: 4.336
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java: 4.428
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableOnErrorResumeNext.java: 4.436
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java: 4.555
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimeout.java: 4.396
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java: 4.190
src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java: 4.525
src/test/java/io/reactivex/rxjava4/core/RxJavaTest.java: 4.537
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java: 4.849
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDelayTest.java: 4.938
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java: 4.761
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupByTest.java: 4.637
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableOnErrorResumeNextTest.java: 4.810

✅ No secrets or suspicious high-entropy strings found.

Mid-4 beauty heuristic in action — powered by our entropy chats! 😊

@codecov

codecov Bot commented Jul 6, 2026

Copy link
Copy Markdown

Codecov Report

❌ Patch coverage is 98.85714% with 2 lines in your changes missing coverage. Please review.
✅ Project coverage is 98.68%. Comparing base (79a67f1) to head (56a477d).

Files with missing lines Patch % Lines
...ex/rxjava4/processors/DispatchStreamProcessor.java 97.84% 1 Missing and 1 partial ⚠️
Additional details and impacted files
@@             Coverage Diff              @@
##                4.x    #8217      +/-   ##
============================================
- Coverage     98.68%   98.68%   -0.01%     
- Complexity     7026     7070      +44     
============================================
  Files           809      810       +1     
  Lines         49714    49855     +141     
  Branches       6719     6749      +30     
============================================
+ Hits          49060    49197     +137     
- Misses          458      460       +2     
- Partials        196      198       +2     

☔ View full report in Codecov by Harness.
📢 Have feedback on the report? Share it here.

🚀 New features to boost your workflow:
  • ❄️ Test Analytics: Detect flaky tests, report on failures, and find test suite problems.

@github-actions

github-actions Bot commented Jul 6, 2026

Copy link
Copy Markdown

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.586 bits/char
Verdict: ✅ Mid-4 beauty detected (thoughtful human code!)

Changed files entropy:

src/main/java/io/reactivex/rxjava4/core/StreamProcessor.java: 4.960
src/main/java/io/reactivex/rxjava4/core/Streamable.java: 4.655
src/main/java/io/reactivex/rxjava4/core/StreamerInput.java: 4.741
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollector.java: 4.407
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDelay.java: 4.580
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableEmpty.java: 4.871
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableForEach.java: 4.316
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupBy.java: 4.336
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableHelper.java: 4.428
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableOnErrorResumeNext.java: 4.436
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTakeUntil.java: 4.555
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableTimeout.java: 4.396
src/main/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableZip.java: 4.190
src/main/java/io/reactivex/rxjava4/internal/util/ExceptionHelper.java: 4.525
src/test/java/io/reactivex/rxjava4/core/RxJavaTest.java: 4.537
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableCollectorTest.java: 4.849
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableDelayTest.java: 4.938
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableFailingFinish.java: 4.761
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableGroupByTest.java: 4.637
src/test/java/io/reactivex/rxjava4/internal/operators/streamable/StreamableOnErrorResumeNextTest.java: 4.810

✅ No secrets or suspicious high-entropy strings found.

Mid-4 beauty heuristic in action — powered by our entropy chats! 😊

@akarnokd akarnokd merged commit db62338 into ReactiveX:4.x Jul 6, 2026
6 checks passed
@akarnokd akarnokd deleted the 4xStreamableProcessors1 branch July 6, 2026 15:45
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment

Projects

None yet

Development

Successfully merging this pull request may close these issues.

1 participant