Adventures with ReactiveStreams

In today's world, more and more people are creating software that is expected to run at large scale. Large scale and cost-effectiveness often lead programmers to asynchronous programming models for handling vertical scalability concerns. Creating an asynchronous application on the JVM is nontrivial and often leads to disparate abstractions. Reactive Streams is one such abstraction that is widely accepted on the JVM today. This presentation is a deep dive into what it takes to implement the Reactive Streams specifications. It delves into various trade-offs that library developers have to consider while implementing Reactive Streams and which aspects of practical applications make such trade-offs easier.

Presented at Oracle Code One 2018:
https://oracle.rainfocus.com/widget/oracle/oow18/catalogcodeone18?search.codeonetracks=1522259325398009d52s&search=DEV5269

Transcript

1. Adventures with Reactive Streams @NiteshKant Nitesh Kant @jovoordeckers Jo Voordeckers
2. Image source: https://interestingengineering.com/12-of-the-worlds-most-fascinating-dams
3. Image source: https://en.wikipedia.org/wiki/Oroville_Dam_crisis
4. @NiteshKant Nitesh Kant @jovoordeckers Jo Voordeckers
5. Find my friends GET /friends?id=3 HTTP/1.1
6. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF …
7. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} …
8. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} …
9. HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() Find my friends getFriendsCsv() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101
10. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
11. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
12. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
13. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
14. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
15. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … Producer Consumer HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send()
16. HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … Producer Consumer Control response.send()
17. I am interested in consuming data! Control
18. I am interested in consuming data! Give me 10 items! Control
19. I am interested in consuming data! Give me 10 items! I am not interested any more. Control
20. Java Message Service https://docs.oracle.com/javaee/5/tutorial/doc/bncdx.html // Synchronous Message m = consumer.receive(); Message m = consumer.receive(1000); // time out 1000ms // Asynchronous MessageListener myListener = new MyListener(); // void onMessage(Message message) consumer.setMessageListener(myListener); request(1)
21. Java Message Service https://docs.oracle.com/javaee/5/tutorial/doc/bncdx.html // Synchronous Message m = consumer.receive(); Message m = consumer.receive(1000); // time out 1000ms // Asynchronous MessageListener myListener = new MyListener(); // void onMessage(Message message) consumer.setMessageListener(myListener); flow control depends on broker
22. Apache Kafka https://kafka.apache.org/10/javadoc/?org/apache/kafka/clients/consumer/KafkaConsumer.html KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); // time out for (ConsumerRecord<String, String> record : records) { // iterate over batch } consumer.commitSync(); } // Flow control public void pause(java.util.Collection<TopicPartition> partitions) public void resume(java.util.Collection<TopicPartition> partitions) request(n)
23. Apache Kafka https://kafka.apache.org/10/javadoc/?org/apache/kafka/clients/consumer/KafkaConsumer.html KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); // time out for (ConsumerRecord<String, String> record : records) { // iterate over batch } consumer.commitSync(); } // Flow control public void pause(java.util.Collection<TopicPartition> partitions) public void resume(java.util.Collection<TopicPartition> partitions) additional flow control by topic/partition
24. Apache Kafka JMS // Synchronous Message m = consumer.receive(); Message m = consumer.receive(1000); // Asynchronous // void onMessage(Message message) MessageListener myListener = new MyListener(); consumer.setMessageListener(myListener); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { // iterate over batch } consumer.commitSync(); } // Flow control public void pause(java.util.Collection<TopicPartition> partitions) public void resume(java.util.Collection<TopicPartition> partitions)
25. Apache Kafka JMS // Synchronous Message m = consumer.receive(); Message m = consumer.receive(1000); // Asynchronous // void onMessage(Message message) MessageListener myListener = new MyListener(); consumer.setMessageListener(myListener); KafkaConsumer<String, String> consumer = new KafkaConsumer<>(props); consumer.subscribe(Arrays.asList("foo", "bar")); while (true) { ConsumerRecords<String, String> records = consumer.poll(100); for (ConsumerRecord<String, String> record : records) { // iterate over batch } consumer.commitSync(); } // Flow control public void pause(java.util.Collection<TopicPartition> partitions) public void resume(java.util.Collection<TopicPartition> partitions) how do we bridge these APIs?
26. Service A Service B
27. Service A Service B cache ?
28. Service A Service B cache Service C ?
29. Service A Service B cache Service C DB ?
30. Service A Service B cache Service C DB ? define contract for interoperability
31. Service A Service B cache Service C DB write adapters for all the things \o/ ?
32. Service A Service B cache Service C DB write adapters for all the things \o/ ? ✓ cancelation ✓ buffering
33. https://github.com/reactive-streams/reactive-streams-jvm
34. https://github.com/reactive-streams/reactive-streams-jvm
35. Reactive Streams public interface Publisher<T> { public void subscribe(Subscriber<? super T> s); }
36. Reactive Streams public interface Publisher<T> { public void subscribe(Subscriber<? super T> s); } I am interested in consuming data!
37. Reactive Streams public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); }
38. Reactive Streams public interface Subscriber<T> { public void onSubscribe(Subscription s); public void onNext(T t); public void onError(Throwable t); public void onComplete(); } Control exchange
39. Reactive Streams public interface Subscription { public void request(long n); public void cancel(); }
40. Reactive Streams public interface Subscription { public void request(long n); public void cancel(); } Give me “n” items!
41. Reactive Streams public interface Subscription { public void request(long n); public void cancel(); } I am not interested any more.
42. under-specification Image source: https://newrepublic.com/article/118416/what-dhaka-bangladesh-traffic-capital-world-can-teach-us
43. under-specification risk of chaos between actors of the interaction rules Image source: https://newrepublic.com/article/118416/what-dhaka-bangladesh-traffic-capital-world-can-teach-us
44. under-specification risk of chaos between actors target least common denominator of the interaction rules Image source: https://newrepublic.com/article/118416/what-dhaka-bangladesh-traffic-capital-world-can-teach-us
45. under-specification risk of chaos between actors target least common denominator systemic inefficiency of the interaction rules Image source: https://newrepublic.com/article/118416/what-dhaka-bangladesh-traffic-capital-world-can-teach-us
46. subscribe() Publisher Subscriber
47. onSubscribe() subscribe() Publisher Subscription onSubscribe() Subscriber
48. request() onSubscribe() subscribe() Publisher Subscription onSubscribe() Subscriber
49. onNext() onComplete() onError() onSubscribe() subscribe() Publisher Subscription onNext() onComplete() onError() onSubscribe() Subscriber request()
50. sequence
51. sequence onSubscribe() onComplete() subscribe()
52. sequence onSubscribe() onComplete() subscribe() request() onSubscribe() subscribe() onError()
53. sequence onSubscribe() onComplete() subscribe() request() onSubscribe() subscribe() onError() request() onSubscribe() subscribe() onNext() onNext() onComplete() request()
54. sequence onSubscribe() onComplete() subscribe() request() onSubscribe() subscribe() onError() request() onSubscribe() subscribe() onNext() onError() request() onSubscribe() subscribe() onNext() onNext() onComplete() request()
55. sequence onSubscribe() onComplete() subscribe() request() onSubscribe() subscribe() onError() request() onSubscribe() subscribe() onNext() onError() request() onSubscribe() subscribe() onNext() onNext() onComplete() request() request() cancel() onSubscribe() subscribe()
56. sequence onSubscribe() onComplete() subscribe() request() onSubscribe() subscribe() onError() request() onSubscribe() subscribe() onNext() onError() request() onSubscribe() subscribe() onNext() onNext() onComplete() request() request() cancel() onSubscribe() subscribe() request() cancel() onNext() onComplete() onSubscribe() subscribe()
57. lifecycle management
58. lifecycle management onNext() X subscribe()
59. lifecycle management onNext() X subscribe() onNext() onSubscribe() onNext() request() onComplete() X onError() subscribe() X
60. lifecycle management onNext() X subscribe() cancel() request() subscribe() onSubscribe() onNext() request() X onNext() onSubscribe() onNext() request() onComplete() X onError() subscribe() X
61. How difficult could it be to implement?
62. Single item, asynchronously produced and delivered
63. Single item, asynchronously produced and delivered (Attempt #1) class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } } * This is a hot/eager source, i.e. the work has started before a Subscriber arrives
64. Single item, asynchronously produced and delivered (Attempt #1) class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } } * This is a hot/eager source, i.e. the work has started before a Subscriber arrives emit() subscribe() request() cancel()
65. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Asynchrony emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
66. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Asynchrony emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
67. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Asynchrony emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
68. class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } } Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Asynchrony emit() subscribe() request() cancel()
69. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
70. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
71. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
72. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() X class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
73. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() https://github.com/reactive-streams/reactive-streams-jvm#1.3 class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); s.onComplete(); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
74. Single item, asynchronously produced and delivered (Attempt #1) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() https://github.com/reactive-streams/reactive-streams-jvm#2.7 class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); s.onComplete(); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
75. Single item, asynchronously produced and delivered (Attempt #2) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); s.onComplete(); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
76. Single item, asynchronously produced and delivered (Attempt #2) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives Concurrency emit() subscribe() request() cancel() class SingleValue<T> implements Publisher<T> { private T value; private boolean subscriberReady; private Subscriber<? super T> subscriber; @Override public synchronized void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { if (value != null) { s.onNext(value); s.onComplete(); } else { subscriberReady = true; } } @Override public void cancel() { subscriberReady = false; } }); } public synchronized void emit(T value) { if (subscriber != null && subscriberReady) { subscriber.onNext(value); } else { this.value = value; } } }
77. Cost of synchronized? Image source: https://www.shutterstock.com/image-photo/iceberg-above-underwater-view-taken-greenland-1127373485
78. Cost of synchronized? Image source: https://www.shutterstock.com/image-photo/iceberg-above-underwater-view-taken-greenland-1127373485
79. s.onNext(value); public synchronized void emit(T value) ¯\_(ツ)_/¯
80. Single item, asynchronously produced and delivered (Attempt #3) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives https://github.com/reactive-streams/reactive-streams-jvm#3.4
81. class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } Single item, asynchronously produced and delivered (Attempt #4) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives State Machine IDLE EMITTED REQUESTED
82. Single item, asynchronously produced and delivered (Attempt #4) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives State Machine IDLE class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } EMITTED REQUESTED
83. Single item, asynchronously produced and delivered (Attempt #5) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives State Machine IDLE class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int CANCELLED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(CANCELLED); } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } EMITTED REQUESTED CANCELLED
84. class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int CANCELLED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(CANCELLED); } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } Single item, asynchronously produced and delivered (Attempt #5) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives State Machine IDLE EMITTED REQUESTED CANCELLED
85. Single item, asynchronously produced and delivered (Attempt #5) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives subscription.cancel(); subscription.request(1) class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int CANCELLED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(CANCELLED); } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } }
86. Single item, asynchronously produced and delivered (Attempt #6) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives State Machine IDLE class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { subscriber = s; s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndSet(REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { } }); } public void emit(T value) { this.value = value; if (state.getAndSet(EMITTED) == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } EMITTED REQUESTED https://github.com/reactive-streams/reactive-streams-jvm#3.6
87. Single item, asynchronously produced and delivered (Attempt #6) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int CANCELLED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndUpdate(val -> val == CANCELLED ? CANCELLED : REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(CANCELLED); } }); subscriber = s; } public void emit(T value) { this.value = value; int old = state.getAndUpdate(val -> val == CANCELLED ? CANCELLED : EMITTED); if (old == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } }
88. class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int CANCELLED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { s.onSubscribe(new Subscription() { @Override public void request(long n) { int old = state.getAndUpdate(val -> val == CANCELLED ? CANCELLED : REQUESTED); if (old == EMITTED) { s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(CANCELLED); } }); subscriber = s; } public void emit(T value) { this.value = value; int old = state.getAndUpdate(val -> val == CANCELLED ? CANCELLED : EMITTED); if (old == REQUESTED) { subscriber.onNext(value); subscriber.onComplete(); } } } Single item, asynchronously produced and delivered (Attempt #6) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives
89. Single item, asynchronously produced and delivered (Attempt #6) * This is a hot/eager source, i.e. the work has started before a Subscriber arrives class SingleValue<T> implements Publisher<T> { private static final int IDLE = 0; private static final int REQUESTED = 1; private static final int EMITTED = 2; private static final int TERMINATED = 3; private T value; private Subscriber<? super T> subscriber; private final AtomicInteger state = new AtomicInteger(IDLE); @Override public void subscribe(Subscriber<? super T> s) { s.onSubscribe(new Subscription() { @Override public void request(long n) { if (n <= 0 && state.getAndSet(TERMINATED) != TERMINATED) { s.onError(new IllegalArgumentException("Non positive request signals are illegal.")); return; } int old = state.getAndUpdate(val -> val == TERMINATED ? TERMINATED : REQUESTED); if (old == EMITTED) { state.set(TERMINATED); s.onNext(value); s.onComplete(); } } @Override public void cancel() { state.set(TERMINATED); } }); subscriber = s; } public void emit(T value) { this.value = value; int old = state.getAndUpdate(val -> val == TERMINATED ? TERMINATED : EMITTED); if (old == REQUESTED) { state.set(TERMINATED); subscriber.onNext(value); subscriber.onComplete(); } } } Want to see how multiple item emission will look like?
90. Let’s get real … ✋
91. response.send() Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 011101100 101110011 101100101 110011100 101110111 011001101 HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} …
92. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} …
93. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher
94. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher Control
95. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher subscription.request(1)
96. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher subscription.request(1) Demand is required to compose through multiple components
97. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher subscription.cancel() Cancellation is required to compose through multiple components
98. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher Publisher client.get("friends?id=1") response.send(Publisher resp)
99. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher response.send( client.get("friends?id=1") );
100. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher List<Friend> transformToJson(List<String> csvs) { List<Friend> friends; for (String csv : csvs) { friends.add(toJson(csv)); } return friends; }
101. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher response.send( client.get("friends?id=1") );
102. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher response.send( client.get("friends?id=1") .map(csv -> toJson(csv)) );
103. Find my friends GET /friends?id=3 HTTP/1.1 getFriendsCsv() transformToJson() HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … response.send() HTTP/1.1 200 OK Nitesh, Kant, SF Jo, Voordeckers, SF … HTTP/1.1 200 OK {First: Nitesh, Last: Kant, Location: SF} {First: Jo, Last: Voordeckers, Location: SF} … Socket Socket Subscriber Publisher response.send( client.get("friends?id=1") .map(csv -> toJson(csv)) ); Function composition.
104. Reactive Streams is not sufficient as an end-user API.
105. Function Composition Lack of standards
106. Function Composition Lack of standards Project Reactor RxJava JDK Streams Akka Streams Kafka Streams
107. Practically, most applications have blocking code.
108. https://github.com/reactive-streams/reactive-streams-jvm#2.2
109. https://github.com/reactive-streams/reactive-streams-jvm#2.2 Embracing this reality is necessary…
110. Not everything is a stream Publisher<User> getUser(String id);
111. Not everything is a stream Publisher<User> getUser(String id); Publisher<Void> fireAndForget(String log);
112. Not everything is a stream Publisher<User> getUser(String id); Publisher<Void> fireAndForget(String log); Publisher<HttpResponse> request(HttpRequest request);
113. Reactive Streams HTTP Kafka Cassandra RPC Redis ()
114. Reactive Streams HTTP Kafka Cassandra RPC Redis () Function Composition Not everything is a stream Blocking safety
115. Image source: http://sciencemeetsfood.org/team-work/
116. Thank You! @NiteshKant Nitesh Kant @jovoordeckers Jo Voordeckers +