Macro-fusion in primitive operators

[OlegDokuka]

Expected behavior

Fusion should happen when a few identical primitive operators appear in the line

For example:

Flux.range(1, n)
    .map(a -> a + 1)
    .map(a -> a + 1)
    .map(a -> a + 1)
    .map(a -> a + 1)
    .map(a -> a + 1)
    .map(a -> a + 1)

should create only two instances of Flux and 6 + 5 (after applying Function#compose) lambdas.

Lambda should fuse to one over the Function#compose as in the following sample:

function1.compose(function2).compose(functionN)

Note

Fusion should not happen when FluxAssamble appears and provides features such a debug information collecting

Actual behavior

Fusion does not happen because not implemented yet

Steps to reproduce

Reactor Core version

any

JVM version (e.g. java -version)

any

Additional Proposal

Fuse primitive operators that have different behaviors but are primitive by nature.

For example:

Flux.range(1, n)
    .map(a -> b)
    .filter(b -> b % 3) 

can be fused to one

Flux.range(1, n)
    .<Integer>handle((a, sink) -> {
        var result = map.apply(a);
        if (filter.test(result)) {
            sink.next(result);
        }
    }) 

Operators that should be macro-fusible

• [ ] .map -> .map ...
• [ ] .filter -> .filter
• [ ] .handle -> .handle
• [ ] FluxPeek (internal doOnSignal)
• [ ] .doOnEach
• [ ] ??? missed operator here

to consideration

• [ ] .map -> .filter and .filter -> map fuses to .handle
• [ ] .handle -> .filter and .filter -> handle fuses to .handle
• [ ] .handle -> .map and .map -> handle fuses to .handle

[bsideup]

Thing to keep in mind: Hooks.onEachOperator.

[OlegDokuka]

Right. I guess in that case all fusions will be negated

(mentioned in Note section about FluxAssamble)

[simonbasle]

IMO, you're pushing it too far for now with filter, map, handle and doOnEach, better stick with doOn* subset covered by FluxPeek for now...

[simonbasle]

FYI I've started toying around with the idea of macro fusion for map, filter and no FluxPeek (doOnSubscribe|Next|Complete|Error|Terminate|Request|Cancel and doAfterTerminate).

⚠️ For the later, something to keep in mind: when combining similar handlers, if one throws then the other handlers are skipped.

This is congruent with the current state for most handlers (eg. if first of 2 doOnNext throws, then the second doOnNext is never invoked since the FluxPeek#onNext will catch the exception and propagate through onError)....

Except for doOnError!So macro-fusion needs to keep user-defined error handlers notified of exceptions in a cascading fashion. Eg. if upon receiving exception errorOriginal, error-handler 1 throws error1 then error handler 2 get notified of error1-suppressing-errorOriginal.

[chemicL]

Closing due to inactivity for nearly 4 years.