What is Reactive?

we want systems that are Responsive, Resilient, Elastic and Message Driven. We call these Reactive Systems.

What is Reactive?

Fault Tolerant

Backpressure

What about ...?

Why Reactive?

Reactive Libraries

Reactive Enabled Libraries

Many

RxJava 1.x

RxJava 2.x

Reactor Core

Akka Streams

Reactive Streams

public interface Subscriber<T> {
    public void onSubscribe(Subscription s);
    public void onNext(T t);
    public void onError(Throwable t);
    public void onComplete();
}

public interface Publisher<T> {
    public void subscribe(org.reactivestreams.Subscriber<? super T> s);
}

public interface Subscription {
    public void request(long n);
    public void cancel();
}

public interface Processor<T, R> extends Subscriber<T>, Publisher<R> {
}

Reactive Streams

Why Reactive Streams?

Interoperability

Reactive Streams TCK

Marbles

Akka Runnable Graph

Akka Runnable Graph

  final Outlet<Integer> A = builder.add(Source.single(0)).out();
  final UniformFanOutShape<Integer, Integer> B = builder.add(Broadcast.create(2));
  final UniformFanInShape<Integer, Integer> C = builder.add(Merge.create(2));
  final FlowShape<Integer, Integer> D =
    builder.add(Flow.of(Integer.class).map(i -> i + 1));
  final UniformFanOutShape<Integer, Integer> E = builder.add(Balance.create(2));
  final UniformFanInShape<Integer, Integer> F = builder.add(Merge.create(2));
  final Inlet<Integer> G = builder.add(Sink.<Integer> foreach(System.out::println)).in();

  builder.from(F).toFanIn(C);
  builder.from(A).viaFanOut(B).viaFanIn(C).toFanIn(F);
  builder.from(B).via(D).viaFanOut(E).toFanIn(F);
  builder.from(E).toInlet(G);

• Rx Java 1.x
  Observable.range(1, 100).map(i -> i + 33) .subscribe(System.out::println);
• Rx Java 2.x
  Flowable.range(1, 100).map(i -> i + 33) .subscribe(System.out::println);
• Reactor 3.x
  Flux.range(1, 100).map(i -> i + 33) .subscribe(System.out::println);
• Akka Streams
  Source.range(1, 100).map(i -> i + 33) .runForeach(System.out::println, materializer);

Create Source RxJava 1.x

        Observable.just(8);
        Observable.from(new Integer[]{8});
        Observable.fromCallable(() -> 8);
        Observable.empty();
        Observable.never();
        Observable.error(new IllegalStateException("why 8"));
        Observable.create(subscriber -> {
            subscriber.onNext(8);
            subscriber.onCompleted();
        });
    

Create Source RxJava 2.x

Observable.just(8);
Observable.fromArray(new Integer[]{8});
Observable.fromCallable(() -> 8);
Observable.fromFuture(new IntFuture(8));
Observable.fromIterable(Arrays.asList(new Integer[]{8}));
Observable.fromPublisher(new IntPublisher(8));
Observable.empty();
Observable.never();
Observable.error(new IllegalStateException("why 8"));
Observable.create(e -> {
    e.setCancellable(() -> {});
    e.onNext(8);
    e.onComplete();});
    

Flowable.just(8);
Flowable.fromArray(8);
Flowable.fromCallable(() -> 8);
Flowable.fromFuture(new IntFuture(8));
Flowable.fromIterable(Arrays.asList(new Integer[]{8}));
Flowable.fromPublisher(new IntPublisher(8));
Flowable.empty();
Flowable.never();
Flowable.error(new IllegalStateException("why 8"));
Flowable.create(emitter -> {
    emitter.setCancellable(() -> {});
    emitter.onNext(8);
    emitter.onComplete();}, BackpressureStrategy.BUFFER);
    

Create Source Reactor

        Flux.just(8);
        Flux.fromArray(new Integer[]{8});
        Flux.from(new IntPublisher(8));
        Flux.fromIterable(Arrays.asList(new Integer[]{8}));
        Flux.fromStream(Stream.of(8));
        Flux.empty();
        Flux.never();
        Flux.error(new IllegalStateException("why 8"));
        Flux.create(e -> {
            e.onDispose(()->{});
            e.next(8);
            e.complete();
        }, FluxSink.OverflowStrategy.BUFFER);
    

Testing RxJava

TestSubscriber<Integer> ts = Flowable.range(1, 5).test();

TestObserver<Integer> to = Observable.range(1, 5).test();

TestObserver<Integer> tso = Single.just(1).test();

TestObserver<Integer> tmo = Maybe.just(1).test();

TestObserver<Integer> tco = Completable.complete().test();
            

Fusion?

• many sequences are started from constant or quasi-constant sources such as just(), from(T[]), from(Iterable), fromCallable() which don't really need the thread-safety dance in a sequence of operators
• some pairs of operators can share internal components such as Queues
• some operators can tell if they consumed the value or dropped it, avoiding request(1) call overhead

Macro Fusion

• Replacing an operator with another operator
• Replacing an operator with custom operator
• Replacing during subscription time
• Replacing with the same operator but with different params

Micro Fusion

• Conditional Subscriber
• Synchronous-fusion
• Asynchronous-fusion

Don't Block, React !