Reactive Streams with Rx at JavaOne 2014

Learn how Netflix applies the Rx programming model to streams of data in finite, infinite and distributed infinite stream uses cases including a distributed system doing anomaly detection.

The presentation digs into the details of flow control and various approaches to achieve it including "reactive pull", a dynamic push/pull approach that composes backpressure into the streams.

Presented at JavaOne 2014: https://oracleus.activeevents.com/2014/connect/sessionDetail.ww?SESSION_ID=5749

Video: http://www.parleys.com/play/543f8d5be4b06e1184ae4106

More information can be found at http://reactivex.io

Transcript

1. Ben Christensen Developer – Edge Engineering at Netflix @benjchristensen ! ! ! ! ! ! http://techblog.netflix.com/ JavaOne - September 2014 Reactive Streams with Rx
3. RxJava http://github.com/ReactiveX/RxJava http://reactivex.io
5. Single Multiple Sync T getData() Iterable<T> getData() Async Future<T> getData() Observable<T> getData()
6. public interface Observer<T> { public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Observer<? super T> subscriber) } Lazy Cold or Hot Asynchronous Push *onNext (onError | onCompleted)?
7. public interface Observer<T> { public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Observer<? super T> subscriber) } Lazy Cold or Hot Asynchronous (or Synchronous) Push (or Pull) *onNext (onError | onCompleted)?
8. public interface Observer<T> { public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Observer<? super T> subscriber) } Lazy Cold or Hot Asynchronous (or Synchronous) Push (or Pull) *onNext (onError | onCompleted)?
9. public interface Subscriber<T> implements Observer<T> { public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) } Lazy Cold or Hot Asynchronous (or Synchronous) Push (or Pull) *onNext (onError | onCompleted)?
10. public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
11. public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
12. public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
13. public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
14. public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } “reactive pull” dynamic push/pull for flow control (backpressure)
15. public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) } public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
16. public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) } public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
17. public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) } public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); }
18. Observable.create(subscriber -> { subscriber.onNext("Hello world!"); subscriber.onCompleted(); }).forEach(System.out::println); synchronous single value public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
19. Observable.create(subscriber -> { subscriber.onNext("Hello"); subscriber.onNext("world"); subscriber.onNext("!"); subscriber.onCompleted(); }).forEach(System.out::println); synchronous multiple values public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
20. Observable.create(subscriber -> { try { subscriber.onNext(doSomething()); subscriber.onCompleted(); } catch (Throwable e) { subscriber.onError(e); } }).subscribeOn(Schedulers.io()) .forEach(System.out::println); asynchronous single value error notification public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
21. Observable.create(subscriber -> { int i = 0; while (!subscriber.isUnsubscribed()) { subscriber.onNext(i++); } }).take(10).forEach(System.out::println); synchronous multiple values with unsubscribe public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
22. Observable.create(subscriber -> { AtomicInteger i = new AtomicInteger(); AtomicLong requested = new AtomicLong(); subscriber.setProducer(r -> { if (requested.getAndAdd(r) == 0) { do { if (subscriber.isUnsubscribed()) { break; } subscriber.onNext(i.incrementAndGet()); } while (requested.decrementAndGet() > 0); } }); }).observeOn(Schedulers.newThread()) .take(10).forEach(System.out::println); synchronous multiple values flow control with “reactive pull” backpressure thread scheduling public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
23. Observable.create(subscriber -> { AtomicInteger i = new AtomicInteger(); AtomicLong requested = new AtomicLong(); subscriber.setProducer(r -> { if (requested.getAndAdd(r) == 0) { do { if (subscriber.isUnsubscribed()) { break; } subscriber.onNext(i.incrementAndGet()); } while (requested.decrementAndGet() > 0); } }); }).observeOn(Schedulers.newThread()) .take(10).forEach(System.out::println); public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
24. Observable.from(iterable) .observeOn(Schedulers.newThread()) .take(10).forEach(System.out::println); public static interface OnSubscribe<T> { public void call(Subscriber<? super T> subscriber); } public abstract class Subscriber<T> implements Observer<T>, Subscription public abstract void onCompleted() public abstract void onError(Throwable e) public abstract void onNext(T t) ! public final void add(Subscription s) public void setProducer(Producer producer) } ! public interface Producer { public void request(long n); } public class Observable<T> { public static <T> Observable<T> create(OnSubscribe<T> f) public Subscription subscribe(Subscriber<? super T> subscriber) }
25. finite
26. infinite finite
27. finite infinite distributed
32. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
33. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
34. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
35.  Observable<R>  b  =  Observable<T>.flatMap({  T  t  -­‐>            Observable<R>  r  =  ...  transform  t  ...          return  r;    }) flatMap
36.  Observable<R>  b  =  Observable<T>.flatMap({  T  t  -­‐>            Observable<R>  r  =  ...  transform  t  ...          return  r;    }) flatMap
37. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
38. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
39. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
40. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
41. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
42. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
43. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
44. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
45. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
46.        Observable.zip(a,  b,  {  a,  b,  -­‐>                ...  operate  on  values  from  both  a  &  b  ...              return  [a,  b];  //  i.e.  return  tuple          })
47. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
48. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
50. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
51. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
52. class  VideoService  {        def  Observable<VideoList>  getPersonalizedListOfMovies(userId);        def  Observable<VideoBookmark>  getBookmark(userId,  videoId);        def  Observable<VideoRating>  getRating(userId,  videoId);        def  Observable<VideoMetadata>  getMetadata(videoId);   } an observable api: finite Treat everything like a stream
53. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); } 1 2 3 4 5 6
54. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); } 3 4 5
55. public Observable<Void> handle(HttpServerRequest<ByteBuf> request, HttpServerResponse<ByteBuf> response) { // first request User object return new UserCommand(request.getQueryParameters().get("userId")).observe().flatMap(user -> { // then fetch personal catalog Observable<Map<String, Object>> catalog = new PersonalizedCatalogCommand(user).observe() .flatMap(catalogList -> { return catalogList.videos().<Map<String, Object>> flatMap(video -> { Observable<Bookmark> bookmark = new BookmarkCommand(video).observe(); Observable<Rating> rating = new RatingsCommand(video).observe(); Observable<VideoMetadata> metadata = new VideoMetadataCommand(video).observe(); return Observable.zip(bookmark, rating, metadata, (b, r, m) -> { return combineVideoData(video, b, r, m); }); }); }); // and fetch social data in parallel Observable<Map<String, Object>> social = new SocialCommand(user).observe().map(s -> { return s.getDataAsMap(); }); // merge the results return Observable.merge(catalog, social); }).flatMap(data -> { // output as SSE as we get back the data (no waiting until all is done) return response.writeAndFlush(new ServerSentEvent(SimpleJson.mapToJson(data))); }); }
56. infinite
57. private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingPivot(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> instanceStreams) { return instanceStreams.map(instanceStream -> { return createGroupedObservable(instanceStream.getKey(), instanceStream .groupBy((Map<String, Object> json) -> { return TypeAndNameKey.from(String.valueOf(json.get("type")), String.valueOf(json.get("name"))); })); }).lift(OperatorPivot.create()).map(commandGroup -> { // merge all instances per group into a single stream of deltas and sum them return createGroupedObservable(commandGroup.getKey(), commandGroup.flatMap(instanceGroup -> { return instanceGroup.startWith(Collections.<String, Object> emptyMap()) .buffer(2, 1) .map(StreamAggregator::previousAndCurrentToDelta) .filter(data -> data != null && !data.isEmpty()); }).scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta) .skip(1)); }); }
58. private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingPivot(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> instanceStreams) { return instanceStreams.map(instanceStream -> { return createGroupedObservable(instanceStream.getKey(), instanceStream .groupBy((Map<String, Object> json) -> { return TypeAndNameKey.from(String.valueOf(json.get("type")), String.valueOf(json.get("name"))); })); }).lift(OperatorPivot.create()).map(commandGroup -> { // merge all instances per group into a single stream of deltas and sum them return createGroupedObservable(commandGroup.getKey(), commandGroup.flatMap(instanceGroup -> { return instanceGroup.startWith(Collections.<String, Object> emptyMap()) .buffer(2, 1) .map(StreamAggregator::previousAndCurrentToDelta) .filter(data -> data != null && !data.isEmpty()); }).scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta) .skip(1)); }); }
60. private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingPivot(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> instanceStreams) { return instanceStreams.map(instanceStream -> { return createGroupedObservable(instanceStream.getKey(), instanceStream .groupBy((Map<String, Object> json) -> { return TypeAndNameKey.from(String.valueOf(json.get("type")), String.valueOf(json.get("name"))); })); }).lift(OperatorPivot.create()).map(commandGroup -> { // merge all instances per group into a single stream of deltas and sum them return createGroupedObservable(commandGroup.getKey(), commandGroup.flatMap(instanceGroup -> { return instanceGroup.startWith(Collections.<String, Object> emptyMap()) .buffer(2, 1) .map(StreamAggregator::previousAndCurrentToDelta) .filter(data -> data != null && !data.isEmpty()); }).scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta) .skip(1)); }); }
62. private static Observable<GroupedObservable<TypeAndNameKey, Map<String, Object>>> aggregateUsingPivot(Observable<GroupedObservable<InstanceKey, Map<String, Object>>> instanceStreams) { return instanceStreams.map(instanceStream -> { return createGroupedObservable(instanceStream.getKey(), instanceStream .groupBy((Map<String, Object> json) -> { return TypeAndNameKey.from(String.valueOf(json.get("type")), String.valueOf(json.get("name"))); })); }).lift(OperatorPivot.create()).map(commandGroup -> { // merge all instances per group into a single stream of deltas and sum them return createGroupedObservable(commandGroup.getKey(), commandGroup.flatMap(instanceGroup -> { return instanceGroup.startWith(Collections.<String, Object> emptyMap()) .buffer(2, 1) .map(StreamAggregator::previousAndCurrentToDelta) .filter(data -> data != null && !data.isEmpty()); }).scan(new HashMap<String, Object>(), StreamAggregator::sumOfDelta) .skip(1)); }); }
64. data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":317,"rollingCountSuccess":327,"reportingHosts":1} data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":1376,"rollingCountSuccess":1432,"reportingHosts":4}
65. data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":317,"rollingCountSuccess":327,"reportingHosts":1} data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":1376,"rollingCountSuccess":1432,"reportingHosts":4}
66. data => {"rollingCountFallbackFailure":0,"rollingCountFallbackSuccess": 0,"propertyValue_circuitBreakerRequestVolumeThreshold":"20","propertyValue_circuitBreakerForceOpen":false,"propertyValue_metricsRol lingStatisticalWindowInMilliseconds":"10000","latencyTotal_mean":32,"type":"HystrixCommand","rollingCountResponsesFromCache": 0,"rollingCountTimeout":0,"propertyValue_executionIsolationStrategy":"SEMAPHORE","instanceId":"34567","rollingCountFailure": 0,"rollingCountExceptionsThrown":0,"latencyExecute_mean":32,"isCircuitBreakerOpen":false,"errorCount": 0,"rollingCountSemaphoreRejected":0,"latencyTotal":{"0":0,"25":0,"50":12,"75":48,"90":68,"95":96,"99":192,"99.5":1452,"100": 1560},"requestCount":1376,"rollingCountCollapsedRequests":94516,"rollingCountShortCircuited":0,"latencyExecute":{"0":0,"25":0,"50": 12,"75":48,"90":68,"95":96,"99":192,"99.5":1452,"100": 1560},"propertyValue_circuitBreakerSleepWindowInMilliseconds":"5000","currentConcurrentExecutionCount": 0,"propertyValue_executionIsolationSemaphoreMaxConcurrentRequests":"10","errorPercentage":0,"rollingCountThreadPoolRejected": 0,"propertyValue_circuitBreakerEnabled":true,"propertyValue_executionIsolationThreadInterruptOnTimeout":true,"propertyValue_request CacheEnabled":true,"rollingCountFallbackRejection":0,"propertyValue_requestLogEnabled":true,"rollingCountSuccess": 1432,"propertyValue_fallbackIsolationSemaphoreMaxConcurrentRequests":"10","propertyValue_circuitBreakerErrorThresholdPercentage":"5 0","propertyValue_circuitBreakerForceClosed":false,"name":"GetPredictions","reportingHosts": 4,"propertyValue_executionIsolationThreadPoolKeyOverride":"null","propertyValue_executionIsolationThreadTimeoutInMilliseconds":"100 0"} data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":317,"rollingCountSuccess":327,"reportingHosts":1} data: {“type":"HystrixCommand","name":"GetPredictions","requestCount":1376,"rollingCountSuccess":1432,"reportingHosts":4}
67. Flow Control
68. Flow Control (backpressure)
69. no backpressure needed Observable.from(iterable).take(1000).map(i -> "value_" + i).subscribe(System.out::println);
70. Observable.from(iterable).take(1000).map(i -> "value_" + i).subscribe(System.out::println); no backpressure needed synchronous on same thread (no queueing)
71. Observable.from(iterable).take(1000).map(i -> "value_" + i) .observeOn(Schedulers.computation()).subscribe(System.out::println); backpressure needed
72. Observable.from(iterable).take(1000).map(i -> "value_" + i) .observeOn(Schedulers.computation()).subscribe(System.out::println); backpressure needed asynchronous (queueing)
73. Flow Control Options
74. Block (callstack blocking and/or park the thread)
75. Temporal Operators (batch or drop data using time)
76. Observable.range(1, 1000000).sample(10, TimeUnit.MILLISECONDS).forEach(System.out::println); 110584 242165 544453 942880
77. Observable.range(1, 1000000).throttleFirst(10, TimeUnit.MILLISECONDS).forEach(System.out::println); 1 55463 163962 308545 457445 592638 751789 897159
78. Observable.range(1, 1000000).debounce(10, TimeUnit.MILLISECONDS).forEach(System.out::println); 1000000
79. Observable.range(1, 1000000).buffer(10, TimeUnit.MILLISECONDS) .toBlocking().forEach(list -> System.out.println("batch: " + list.size())); batch: 71141 batch: 49488 batch: 141147 batch: 141432 batch: 195920 batch: 240462 batch: 160410
81. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] []
82. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] []
83. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] []
84. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] []
85. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] []
86. /* The following will emit a buffered list as it is debounced */ // first we multicast the stream ... using refCount so it handles the subscribe/unsubscribe Observable<Integer> burstStream = intermittentBursts().take(20).publish().refCount(); // then we get the debounced version Observable<Integer> debounced = burstStream.debounce(10, TimeUnit.MILLISECONDS); // then the buffered one that uses the debounced stream to demark window start/stop Observable<List<Integer>> buffered = burstStream.buffer(debounced); // then we subscribe to the buffered stream so it does what we want buffered.toBlocking().forEach(System.out::println); [0, 1, 2] [0, 1, 2] [0, 1, 2, 3, 4, 5, 6] [0, 1, 2, 3, 4] [0, 1] [] https://gist.github.com/benjchristensen/e4524a308456f3c21c0b
87. Observable.range(1, 1000000).window(50, TimeUnit.MILLISECONDS) .flatMap(window -> window.count()) .toBlocking().forEach(count -> System.out.println("num items: " + count)); num items: 477769 num items: 155463 num items: 366768
88. Observable.range(1, 1000000).window(500000) .flatMap(window -> window.count()) .toBlocking().forEach(count -> System.out.println("num items: " + count)); num items: 500000 num items: 500000
89. Reactive Pull (dynamic push-pull)
90. Push when consumer keeps up with producer. ! Switch to Pull when consumer is slow. ! Bound all* queues.
91. Push when consumer keeps up with producer. ! Switch to Pull when consumer is slow. ! Bound all* queues. *vertically, not horizontally
92. Observable Subscriber Observable.create(subscriber -> { // emit to subscriber }).subscribe(new Subscriber<String>() { ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { ! } ! });
93. observable.subscribe Observable.create(subscriber -> { // emit to subscriber }).subscribe(new Subscriber<String>() { ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { ! } ! });
94. observable.subscribe subscriber.setProducer Observable.create(subscriber -> { subscriber.setProducer(request -> { }); }).subscribe(new Subscriber<String>() { ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { ! } ! });
95. Observable.create(subscriber -> { subscriber.setProducer(request -> { }); }).subscribe(new Subscriber<String>() { ! public void onStart() { request(5); } ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { ! } ! }); observable.subscribe subscriber.setProducer producer.request(n)
96. Observable.create(subscriber -> { subscriber.setProducer(request -> { subscriber.onNext(t); // 5 times }); }).subscribe(new Subscriber<String>() { ! public void onStart() { request(5); } ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { ! } ! }); observable.subscribe subscriber.setProducer producer.request(n) subscriber.onNext(t) …
97. Observable.create(subscriber -> { subscriber.setProducer(request -> { subscriber.onNext(t); // 5 times }); }).subscribe(new Subscriber<String>() { ! public void onStart() { request(5); } ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { doStuffWith(t); if(requested—- == 0) request(5); } ! }); observable.subscribe subscriber.setProducer producer.request(n) subscriber.onNext(t) … producer.request(n)
98. Observable.create(subscriber -> { subscriber.setProducer(request -> { … previous logic … subscriber.onCompleted(); }); }).subscribe(new Subscriber<String>() { ! public void onStart() { request(5); } ! public void onCompleted() { // do stuff } ! public void onError(Throwable e) { // handle error } ! public void onNext(String t) { doStuffWith(t); if(requested—- == 0) request(5); } ! }); observable.subscribe subscriber.setProducer producer.request(n) subscriber.onNext(t) … producer.request(n) subscriber.onCompleted/onError unsubscribe
99. Observable Subscriber Observable Subscriber observable.subscribe observable.subscribe
100. Observable.from(iterable) .observeOn(Schedulers.computation()) .subscribe(System.out::println);
101. Observable.from(iterable) .observeOn(Schedulers.computation()) .subscribe(System.out::println);
102. Observable Subscriber Observable Subscriber observable.subscribe observable.subscribe
103. Observable Subscriber Observable Subscriber subscriber.setProducer subscriber.setProducer
104. Observable Subscriber Observable Subscriber producer.request(n) producer.request(5) outstanding requested = 5
105. Observable Subscriber Observable Subscriber producer.request(n) producer.request(5) outstanding requested = 5
106. Observable Subscriber Observable Subscriber producer.request(n) producer.request(5) outstanding requested = 5
107. Observable Subscriber Observable Subscriber subscriber.onNext outstanding requested = 5
108. Observable Subscriber Observable Subscriber outstanding requested = 5 subscriber.onNext
109. Observable Subscriber Observable Subscriber outstanding requested = 4 subscriber.onNext
110. Observable Subscriber Observable Subscriber outstanding requested = 3 subscriber.onNext
111. Observable Subscriber Observable Subscriber outstanding requested = 3 subscriber.onNext producer.request(1) request(1)
112. Observable Subscriber Observable Subscriber outstanding requested = 4 subscriber.onNext producer.request(1)
113. Observable Subscriber Observable Subscriber outstanding requested = 4 subscriber.onNext subscriber.onNext
114. Observable Subscriber Observable Subscriber outstanding requested = 0 subscriber.onNext
115. Observable Subscriber Observable Subscriber outstanding requested = 0 subscriber.onNext
116. Observable Subscriber Observable Subscriber outstanding requested = 0 subscriber.onNext
117. Observable Subscriber Observable Subscriber outstanding requested = 0 subscriber.onNext
118. Observable Subscriber Observable Subscriber outstanding requested = 0 subscriber.onNext
119. Observable Subscriber Observable Subscriber outstanding requested = 2 subscriber.onNext request(2) producer.request(2)
120. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { // do something with the emitted item "n" // request another item: request(1); } });
121. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { // do something with the emitted item "n" // request another item: request(1); } });
122. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { // do something with the emitted item "n" // request another item: request(1); } });
123. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1024); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { enqueue(n); // elsewhere the queue is drained // and request(m) is called } ! });
124. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1024); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { enqueue(n); // elsewhere the queue is drained // and request(m) is called } ! });
125. someObservable.subscribe(new Subscriber<T>() { @Override public void onStart() { request(1024); } ! @Override public void onCompleted() { // gracefully handle sequence-complete } ! @Override public void onError(Throwable e) { // gracefully handle error } ! @Override public void onNext(T n) { enqueue(n); // elsewhere the queue is drained // and request(m) is called } ! });
126. Reactive Pull hot vs cold?
127. Reactive Pull … or when source doesn’t support request(n)?
128. hotSourceStream.onBackpressureBuffer().observeOn(aScheduler);
129. hotSourceStream.onBackpressureBuffer().observeOn(aScheduler);
130. hotSourceStream.onBackpressureBuffer().observeOn(aScheduler);
131. hotSourceStream.onBackpressureBuffer().observeOn(aScheduler);
132. hotSourceStream.onBackpressureDrop().observeOn(aScheduler);
133. stream.onBackpressure(strategy).subscribe
134. distributed
135. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
136. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
137. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
138. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
139. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
140. (movieId) movieId=12345 movieId=34567
141. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
142. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
143. Stage 1 Stage 2 groupBy Event Streams sink
144. Stage 1 Stage 2 12345 56789 34567
145. Stage 1 Stage 2 12345 56789 34567
146. Stage 1 Stage 2 12345 56789 34567
147. Stage 1 Stage 2 12345 56789 34567
148. Stage 1 Stage 2 12345 56789 34567 12345 56789 34567
149. Stage 1 Stage 2 12345 56789 34567 12345 56789 34567
150. Stage 1 Stage 2 12345 56789 34567 12345 56789 34567
151. Stage 1 Stage 2 12345 56789 34567 12345 56789 34567
152. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
153. 10mins || 1000 4:40-4:50pm 4:50-5:00pm 5:00-5:07pm (burst to 1000) 5:07-5:17pm
154. 10mins || 1000 4:40-4:50pm 4:50-5:00pm 5:00-5:07pm (burst to 1000) 5:07-5:17pm
155. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
156. flatMap
157. flatMap
158. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
160. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
161. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
162. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
163. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
164. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
165. MantisJob .source(NetflixSources.moviePlayAttempts()) .stage(playAttempts -> { return playAttempts.groupBy(playAttempt -> { return playAttempt.getMovieId(); }) }) .stage(playAttemptsByMovieId -> { playAttemptsByMovieId .window(10,TimeUnit.MINUTES, 1000) // buffer for 10 minutes, or 1000 play attempts .flatMap(windowOfPlayAttempts -> { return windowOfPlayAttempts .reduce(new FailRatioExperiment(playAttemptsByMovieId.getKey()), (experiment, playAttempt) -> { experiment.updateFailRatio(playAttempt); experiment.updateExamples(playAttempt); return experiment; }).doOnNext(experiment -> { logToHistorical("Play attempt experiment", experiment.getId(),experiment); // log for offline analysis }).filter(experiment -> { return experiment.failRatio() >= DYNAMIC_PROP("fail_threshold").get(); }).map(experiment -> { return new FailReport(experiment, runCorrelations(experiment.getExamples())); }).doOnNext(report -> { logToHistorical("Failure report", report.getId(), report); // log for offline analysis }) }) }) .sink(Sinks.emailAlert(report -> { return toEmail(report)})) // anomalies trigger events (simple email here)
166. stream.onBackpressure(strategy?).subscribe
167. stream.onBackpressure(buffer).subscribe
168. stream.onBackpressure(drop).subscribe
169. stream.onBackpressure(sample).subscribe
170. stream.onBackpressure(scaleHorizontally).subscribe
171. finite infinite distributed
172. Single Multiple Sync T getData() Iterable<T> getData() Async Future<T> getData() Observable<T> getData()
173. RxJava http://github.com/ReactiveX/RxJava http://reactivex.io
174. ! Reactive Programming in the Netflix API with RxJava http://techblog.netflix.com/2013/02/rxjava-netflix-api.html ! Optimizing the Netflix API http://techblog.netflix.com/2013/01/optimizing-netflix-api.html ! Reactive Extensions (Rx) http://www.reactivex.io ! Reactive Streams https://github.com/reactive-streams/reactive-streams ! ! ! ! ! Ben Christensen @benjchristensen RxJava https://github.com/ReactiveX/RxJava @RxJava RxJS http://reactive-extensions.github.io/RxJS/ @ReactiveX jobs.netflix.com