4.x: Add virtualCreate and virtualTransform to Flowable

[akarnokd]

ℹ️ You need the special operators below to make RxJava's non-blocking backpressure into virtually blocked backpressure.

create

Creates a Flowable from a generator callback, that can emit via VirtualEmitter, run on an ExecutorService provided by the user and
is suspended automatically upon backpressure. The callback is executed inside the virtual thread thus you can call the usual blocking APIs and get suspensions the same way.

The created Flowable will complete once the callback returns normally or with an error if the callback throws an exception.

try (var scope = Executors.newVirtualThreadPerTaskExecutor()) {
    Flowable.virtualCreate(emitter -> {
        for (int i = 1; i <= 5; i++) {
             emitter.emit(1);
        }
    }, scope)
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1, 2, 3, 4, 5);
}

transform

Transforms each upstream value via a callback that can emit zero or more values for each of those upstream values, run on an ExecutorService provided by the user and is suspended automatically upon backpressure. The callback is executed inside the virtual thread thus you can call the usual blocking APIs and get suspensions the same way.

try (var scope = Executors.newVirtualThreadPerTaskExecutor()) {
    Flowable.range(1, 5)
    .virtualTransform((value, emitter) -> {
        emitter.emit(value);
        emitter.emit(value + 1);
    }, scope)
    .test()
    .awaitDone(5, TimeUnit.SECONDS)
    .assertResult(1, 2, 2, 3, 3, 4, 4, 5, 5, 6);
}

blockingXXX

RxJava uses java.util.concurrent locks and CountDownLatches via its blockingXXX which will automatically work within a virtual thread. Therefore, there is no need for a separate interop operator. Just block.

try (var scope = Executors.newVirtualThreadPerTaskExecutor()) {
    scope.submit(() -> {
        var v = Flowable.just(1)
        .delay(1, TimeUnit.SECONDS)
        .blockingLast();
        System.out.println(v);
    });
}

[github-actions]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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

[github-actions]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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

[akarnokd]

Related: #7763

The scheduler upgrade / rewamp will happen in a separate PR.

[codecov]

Codecov Report

❌ Patch coverage is 94.11765% with 9 lines in your changes missing coverage. Please review.
⚠️ Please upload report for BASE (4.x@a5d0c55). Learn more about missing BASE report.

Files with missing lines	Patch %	Lines
...rnal/virtual/FlowableVirtualTransformExecutor.java	95.29%	1 Missing and 3 partials ⚠️
...nternal/virtual/FlowableVirtualCreateExecutor.java	92.50%	0 Missing and 3 partials ⚠️
...vex/rxjava4/internal/virtual/VirtualResumable.java	90.00%	0 Missing and 2 partials ⚠️

Additional details and impacted files

@@          Coverage Diff           @@
##             4.x    #8014   +/-   ##
======================================
  Coverage       ?   99.61%           
  Complexity     ?     6791           
======================================
  Files          ?      755           
  Lines          ?    47872           
  Branches       ?     6420           
======================================
  Hits           ?    47686           
  Misses         ?       81           
  Partials       ?      105           

☔ View full report in Codecov by Sentry.
📢 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]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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

[github-actions]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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

[github-actions]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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

[github-actions]

🐷 TruffleHog + Entropy Beauty Scan

Average entropy of changed code: 4.753 bits/char
Verdict: ⚠️ Consider review — entropy outside sweet spot

Changed files entropy:

.github/workflows/discord-release-announce.yml: 4.753

✅ No secrets or suspicious high-entropy strings found.

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