What is Observable?

• represents a push based collection
• can be combined with other observables
• monad
• any number of values over any amount of time

T next()

onNext(T)

throws new Exception();

onError(Exception)

return;

onCompleted()

Memory Effectiveness

Stream.of(3, 1, 4, 1, 5, 9, 2)
      .flatMap(x -> Stream.of(x, x))
      .map(x -> x * 2)
      .filter(x -> x > new Random().nextInt(9));

Observable.just(3, 1, 4, 1, 5, 9, 2)
          .flatMap(x -> Observable.just(x, x))
          .map(x -> x * 2)
          .filter(x -> x > new Random().nextInt(9));

• intermediate results => less GC
• 1 by 1 processing, data is only iterated over once
• hot potatoe
• lazy approach, no observer - no processing

Factories

• .just()
• .create()
• .from()
• .range()
• .repeat()
• .interval()
• .timer()
• .empty() / .never() / .throw()
• ...

Combinators for Observable[T]

• .map(f: T -> R): Observable[R]
• .flatMap(f: T -> Observable[R]): Observable[R]
• .filter()
• .zip(), .merge(), .concat(), groupBy(), sample(), etc.

• map ~ projection
• flatMap is more powerful - it can change the number of elements/events
• filter ~ selection
• join ~ zip/merge/concat
• relational algebra

Hot vs Cold

• Cold ones start producing notifications after subscription (default).
• Hot ones don't care about subscribers.

Rx is everywhere (JVM)

• Java

  Observable.interval(1, TimeUnit.SECONDS)
            .subscribe(System.out::println);

• Scala

  Observable.interval(1, TimeUnit.SECONDS)
            .subscribe(println(_))

• Groovy

  Observable.interval(1, TimeUnit.SECONDS)
            .subscribe{println it}

• JRuby

  Observable.interval(1, TimeUnit.SECONDS)
            .subscribe {|val| puts val}

• Clojure

  (->
    (Observable/interval 1 TimeUnit/SECONDS)
    (.subscribe (fn [arg] (println arg))))

Rx is everywhere

• .NET

  Observable.Interval(TimeSpan.FromSeconds(1))
            .Subscribe(x => Console.WriteLine(x));

• C++

  rx::observable<>::interval(std::chrono::seconds(1))
     .subscribe(
       [](long v){std::cout << v;},
       [](){std::cout << "OnCompleted";}
     )

• Python

  Observable.interval(1000)
            .subscribe(lambda x:  print x)

• PHP

  \Rx\Observable::interval(1000, $scheduler)
                ->subscribe(function ($x) { echo $x; });

• Dart, Haskell, Android SDK, Objective-C & Swift, etc.

Rx is everywhere

• Angular 2.x
• ES7
• reactive-streams.org
• ${your_project}

• interop between different reactive system

Testability

it('should filter in only prime numbers', function () {
	var source = hot('-1--2--^-3-4-5-6--7-8--9--|');
	var subs =              '^                  !';
	var expected =          '--3---5----7-------|';

	expectObservable(source.filter(isPrime)).toBe(expected);
	expectSubscriptions(source.subscriptions).toBe(subs);
});

• schedulers
• overloaded combinators

Maturity

• it is very solid

Recap

• Observable ~ any number of values over any amount of time.
• Rx make sense for UI as well as for the server-side.
• Can be done in any language.
• Cold Observables are lazy.

Want to Learn More?

• Principles of Reactive Programming @ coursera.org
• reactivex.io/learnrx
• reactivemanifesto.org

• given by Eric meyer who is also co-author of the reactivemanifesto.