Project Reactor: About Fuseable interface SYNC mode
Tags: java, reactiveprogramming, projectreactor
Published: 2024-04-18 11:04:29

# About *reactor.core.Fuseable* interface

Main reason of introduces Fuseable interface is to speed up data flow in operator chain. [Here](https://stackoverflow.com/questions/57360433/what-is-fuseable-interface-in-reactor-project-for) is the discussion on that.

# Example SYNC mode implementation

Below is an example to display the signal received by implements both Fuseable SYNC mode and traditional Reactive Streams's request and onNext cycle. Project Reactor 3.6.3 is used.

```java
package example;

import org.reactivestreams.Subscription;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import reactor.core.CoreSubscriber;
import reactor.core.Fuseable;

import java.util.Objects;

public class SyncDisplaySubscriber<T> implements CoreSubscriber<T> {

    private static final Logger LOGGER = LoggerFactory.getLogger(SyncDisplaySubscriber.class);
    private Subscription subscription;

    @Override
    public void onSubscribe(Subscription s) {
        LOGGER.info("onSubscribe(): s={}", s);
        subscription = s;
        if (s instanceof Fuseable.QueueSubscription) { // (1)
            @SuppressWarnings("unchecked")
            Fuseable.QueueSubscription<T> qs = (Fuseable.QueueSubscription<T>) s;
            int mode = qs.requestFusion(Fuseable.SYNC); // (2)
            LOGGER.info("onSubscribe(): requestFusion()={}", mode);
            if (mode == Fuseable.SYNC) { // (3)
                drain();
                return;
            }
        }
        s.request(1L);
    }

    @Override
    public void onNext(T t) {
        LOGGER.info("onNext(): t={}", t);
        subscription.request(1L);
    }

    @Override
    public void onError(Throwable t) {
        LOGGER.info("onError()", t);
    }

    @Override
    public void onComplete() {
        LOGGER.info("onComplete()");
    }

    private void drain() {
        @SuppressWarnings("unchecked")
        Fuseable.QueueSubscription<T> qs = (Fuseable.QueueSubscription<T>) subscription;
        try {
            T next;
            do {
                next = qs.poll(); // (4)
                if (Objects.nonNull(next)) {
                    LOGGER.info("drain(): next={}", next); // (5)
                } else {
                    LOGGER.info("drain(): complete"); // (6)
                }
            } while (Objects.nonNull(next));
        } catch (Exception t) {
            LOGGER.info("drain(): error", t); // (7)
        }
    }

}
```

(1): the Subscription from upstream operator must implement Fuseable.QueueSubscription, otherwise Fuseable is not supported.

(2): Using Fuseable.QueueSubscription.requestFusion(Fuseable.SYNC) method to check if SYNC mode is supported or not.

(3): SYNC mode is supported only if the return is Fuseable.SYNC. In SYNC mode it is not necessarily to call Subscription.request(long). If Fuseable.NONE is return, the subscriber will get data by Reactive Streams's request and onNext cycle described by [here](https://github.com/reactive-streams/reactive-streams-jvm).

(4): To get a value by calling Fuseable.QueueSubscription.poll(). SYNC mode means that triggers release a value from source.

(5): If the result is not null, it means that onNext signal is received with that value.

(6): If the result is null, it means that the onComplete signal is received.

(7): May get an exception when calling Fuseable.QueueSubscription.poll(). It means that onError signal is received with that exception.

Another class DisplaySubscriber is nearly the same as SyncDisplaySubscriber, except it implements *org.reactivestreams.Subscriber* interface rather than 
*reactor.core.CoreSubscriber* interface.


```java
// Implementation of this class is equal to SyncDisplaySubscriber except implements org.reactivestreams.Subscriber interface
public class DisplaySubscriber<T> implements Subscriber<T> {
    // ...
}
```

Method Flux.just() returns a FluxArray instance, and its Subscription class ArraySubscription supports Fuseable.SYNC mode. We use this operator to observes the data flow of subscribers.

```java
package example;

import reactor.core.publisher.Flux;

public class FuseableSyncModeTest {

    public static void main(String[] args) {
        Flux<Integer> flux1 = Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        flux1.subscribe(new SyncDisplaySubscriber<>());

        Flux<Integer> flux2 = Flux.just(1, 2, 3, 4, 5, 6, 7, 8, 9, 10);
        flux2.subscribe(new DisplaySubscriber<>());
    }

}
```

The output is

```
10:55:23.904 [main] INFO  example.SyncDisplaySubscriber -- onSubscribe(): s=reactor.core.publisher.FluxArray$ArraySubscription@245b4bdc
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- onSubscribe(): requestFusion()=1
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=1
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=2
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=3
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=4
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=5
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=6
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=7
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=8
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=9
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): next=10
10:55:23.905 [main] INFO  example.SyncDisplaySubscriber -- drain(): complete
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onSubscribe(): s=reactor.core.publisher.StrictSubscriber@3590fc5b
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=1
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=2
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=3
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=4
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=5
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=6
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=7
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=8
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=9
10:55:23.917 [main] INFO  example.DisplaySubscriber -- onNext(): t=10
10:55:23.920 [main] INFO  example.DisplaySubscriber -- onComplete()
```

Please note that signal sent from source and received on subscriber is under the same thread. And if the subscriber contains org.reactivestreams.Subscriber only, it means that the data flow should stick on Reactive Streams's request and onNext cycle.