# Akarnokd - Advanced RxJava Programming deep-dive into RxJava, Reactive-Streams, Project Reactor and Java 9 Flow. This is one page of public article previews, not the complete archive. Follow Next page to continue. Summaries are not the original full articles. ## Java 9 Flow API: Multicasting via a Processor DevFeed: [Java 9 Flow API: Multicasting via a Processor]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-12-14T13:20:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [reactive](), [Streams](), [implementation](), [API]() Tags: [backpressure](), [cancellation](), [concurrency](), [flow](), [flow-api](), [java](), [java-9](), [jdk](), [multicast](), [processor](), [reactive](), [reactive-streams](), [streams](), [subscriber](), [subscription](), [tck]() ### AI overview This tutorial explains how Java 9 Flow API processors can multicast elements from a single upstream source to multiple consumers. It focuses on coordinating subscriber demand, backpressure, dynamic subscription, and cancellation, and introduces MulticastProcessor as the intermediate solution. ### Source excerpt Introduction There are situations when the same elements of a source should be dispatched to multiple consumers. Certainly, if the source supports multiple subscribers and is deterministic (such as our previous async range), one can just instantiate the flow multiple times. However, if the source doesn't support multiple subscribers or each subscription ends up being unique and/or non-deterministic, that simple approach doesn't work anymore. We need a way to have a single realization of the (upstream) source yet allow multiple consumers. Since we are dealing with Flow.Publishers that require backpressure management, such intermediate solution has to coordinate requests from its Flow.Subscribers in addition to handling the dynamic subscription and unsubscription (cancellation) of said Flow.Subscribers while the flow is active. Enter, MulticastProcessor. Flow.Processor recap What is a Processor? By definition, it is a combination of a Flow.Publisher and a Flow.Subscriber, i.e., it can act as a source and can be subscribed to via subscribe() as well as the processor itself can be used with somebody else's Flow.Publisher.subscribe(). It has a mixed history as the idea comes from the original Observer pattern (i.e., java.util.Observable) and Rx.NET's Subject that allows dispatching signals to multiple Observers in an imperative (and synchronous) fashion. The Flow.Processor in Java 9 defines two type arguments, one for its input side (Flow.Subscriber) and one for its output side (Flow.Publisher). The idea behind it was that a Flow.Processor can act as a transformation step between an upstream and a downstream. However, such transformation often mandates the Flow.Processor implementation only accepts a single Flow.Subscriber during its entire lifetime. Since the implementation has to follow the Reactive Streams specification nonetheless, this adds a lot of overhead to the flow. As demonstrated in previous posts, when a flow is realized, there are only one subscriber per st ## When multiple subscribeOn()s do have effect DevFeed: [When multiple subscribeOn()s do have effect]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-11-29T12:33:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Code](), [Concurrent Programming]() Tags: [collect](), [concurrency](), [filter](), [flowable](), [io](), [main-thread](), [map](), [scheduler](), [subscribeon](), [take](), [thread](), [tutorials]() ### AI overview This article explains why multiple subscribeOn() operators can sometimes have observable effects. It distinguishes source operators that perform subscription side effects from instance operators that mainly subscribe upstream, and shows how different schedulers can determine the threads where those effects occur. ### Source excerpt Introduction In many tutorials and explanations, it has been said that having multiple subscribeOn()s has no effect and only the one closest to the source wins. I often tell this with the wording "no practical effect". However, it is possible to demonstrate the effects of multiple subscibeOn()s that have some actual effects. What is subscribeOn again? The most precise definition of this operator I can formulate is as follows: subscribeOn changes where (on what thread) the (side) effects of calling subscribe() on the parent/upstream Observable (Flowable, Single, etc.) happen. So what are these subscription (side) effects look like in code? Observable.create(emitter -> { for (int i = 0; i < 10; i++) { emitter.onNext(i + ": " + Thread.currentThread().getName()); } emitter.onComplete(); }) .subscribeOn(Schedulers.io()) .blockingSubscribe(System.out::println); // Prints: // ------- // 0: RxCachedThreadScheduler-1 // 1: RxCachedThreadScheduler-1 // 2: RxCachedThreadScheduler-1 // 3: RxCachedThreadScheduler-1 // 4: RxCachedThreadScheduler-1 // 5: RxCachedThreadScheduler-1 // 6: RxCachedThreadScheduler-1 // 7: RxCachedThreadScheduler-1 // 8: RxCachedThreadScheduler-1 // 9: RxCachedThreadScheduler-1 In this example, the effect of subscribing is that the body of the ObservableOnSubscribe starts running on the thread provided via the io() Scheduler. Applying yet another subscribeOn after the first one won't change what is printed to the console. Most source-like operators, such as create(), fromCallable(), fromIterable(), do have subscription side-effects as they often start emitting event(s) immediately. Most instance operators, such as map(), filter(), take(), don't have subscription side-effects on their own and just subscribe() to their upstream. Instance operators with subscription side-effects However, there are a couple of instance operators that do have subscription side-effects. Specifically, any operator that offers a way to specify a per subscriber initial state via ## Android LiveData API: a quick look DevFeed: [Android LiveData API: a quick look]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-10-19T13:45:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Android](), [API](), [reactive](), [Library](), [Streams]() Tags: [android](), [api](), [backpressure](), [cancellation](), [concurrency](), [flow](), [flowable](), [google](), [lifecycle](), [lifecycle-components](), [livedata](), [reactive](), [release](), [rxjava](), [subscription](), [thread](), [threading](), [ui]() ### AI overview This article provides a quick technical overview of Android LiveData, explaining its main-thread requirements, lifecycle-aware observer behavior, observer removal, and interoperability with Reactive Streams. It notes that LiveData was considered beta and could change before release. ### Source excerpt Introduction Threading and lifecycle are one of the top concerns when developing applications for the Android platform. UI has to be interacted with on a dedicated thread (main thread) but in order to keep the UI responsible to user input and rendering, blocking or CPU intensive calculations should be kept off it. In addition, views can get destroyed and recreated in a way that is outside of a given application's control unlike a desktop Swing application. This means background tasks must be stopped and listeners removed to prevent leaking references to now logically dead objects. RxJava and RxAndroid can help with threading concerns and there are other libraries that tap into the various lifecycle events; in general, this means someone will call dispose() on a particular flow or clear() on a CompositeDisposable to mass-cancel multiples of them. Having a rich set of transformative and coordinating operators along with support for normal values, errors and finite sequences may be overwhelming compared to a classical Listener-based API. Google's LiveData is one of such classical Listener style APIs but unlike Swing's ActionListener for example, there are explicit requirements that interaction with the LiveData object itself happens on the main thread and signals will be dispatched from the main thread to Observers to it. LiveData API Unfortunately, I wasn't able to locate a public repository for the LiveData sources and had to rely on the sources downloaded from Google's Maven repository: compile "android.arch.lifecycle:reactivestreams:1+" There is an interoperation library associated with LiveData that allows presenting and consuming events from any Reactive-Streams Publisher. This will transitively import the actual LiveData library. Note that LiveData is currently considered beta and may change arbitrarily before release. That said, I don't think the core structure and premise will actually change. The main consumer type is the android.arch.lifecycle.Observer with ## Java 9 Flow API: taking and skipping DevFeed: [Java 9 Flow API: taking and skipping]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-30T20:44:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java 9](), [API](), [reactive]() Tags: [api](), [flow](), [flow-api](), [implementation](), [java-9](), [jdk](), [publisher](), [reactive](), [reactive-streams](), [skip](), [skipwhile](), [streams](), [subscriber](), [subscription](), [take](), [takeuntil](), [takewhile]() ### AI overview A tutorial on implementing take and skip-style operators with Java 9's Flow API. It explains how to limit a flow, cancel the upstream subscription when the limit is reached, complete the downstream subscriber, handle terminal events, and account for backpressure behavior. ### Source excerpt Introduction Limiting or skipping over parts of a flow is a very common task: either we are only interested in the first N items or we don't care about the first N items. Sometimes, N is unknown but we can decide, based on the current item, when to stop relaying items or, in contrast, when to start relaying items. Take(N) In concept, limiting a flow to a certain size should be straightforward: count the number of items received via onNext and when the limit is reached, issue a cancel() towards the upstream and onComplete() towards the downstream. public static Flow.Publisher take(Flow.Publisher source, long n) { return new TakePublisher<>(source, n); } The operator's implementation requires little state: static final class TakeSubscriber implements Flow.Subscriber { final Flow.Subscriber downstream; Flow.Subscription upstream; long remaining; TakeSubscriber( Flow.Subscriber downstream, long n) { this.downstream = downstream; this.remaining = n; } @Override public void onSubscribe(Flow.Subscription s) { // TODO implement } @Override public void onNext(T item) { // TODO implement } @Override public void onError(Throwable throwable) { // TODO implement } @Override public void onComplete() { // TODO implement } } In its simplest form, there is no need for intercepting the request() and cancel() calls from the downstream: these can be passthrought, however, since the operator has to stop the sequence upon reaching the limit (remaining == 0), the upstream's Flow.Subscriber has to be stored. @Override public void onSubscribe(Flow.Subscription s) { this.upstream = s; downstream.onSubscribe(s); } In onSubscribe, we only have to store the Flow.Subscription and forward it to the downstream. @Override public void onNext(T item) { long r = remaining; if (r > 0L) { remaining = --r; downstream.onNext(item); if (r == 0) { upstream.cancel(); downstream.onComplete(); } } } While remaining is positive, we decrement it and save it into its field foll ## Java 9 Flow API: arbitration and concatenation DevFeed: [Java 9 Flow API: arbitration and concatenation]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-27T12:55:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java 9](), [Java](), [API]() Tags: [api](), [arbiter](), [arbitration](), [concat](), [concatenation](), [concurrent](), [event](), [flow](), [flow-api](), [java](), [java-9](), [jdk](), [onerrorresumenext](), [publisher](), [repeat](), [request](), [retry](), [submissionpublisher](), [subscriber](), [subscription](), [subscriptionarbiter]() ### AI overview This tutorial explains how to implement subscription arbitration for Java 9's Flow API when concatenating multiple publishers. It addresses stack growth, remaining-demand accounting, concurrent requests and cancellation, subscription switching, and produced-item tracking. ### Source excerpt Introduction A very common task is to combine multiple sources, or more generally, start consuming a source once the previous source has terminated. The naive approach would be to simply call otherSource.subscribe(nextSubscriber) from onError or onComplete. Unfortunately, this doesn't work for two reasons: 1) it may end up with deep stacks due to a "tail" subscription from onError/onComplete and 2) we should request the remaining, unfulfilled amount from the new source that hasn't be provided by the previous source to not overflow the downstream. The first issue can be solved by applying a heavyweight observeOn in general and implementing a basic trampolining loop only for certain concrete cases such as flow concatenation to be described in this post. The second issue requires a more involved source: not only do we have to switch between Flow.Subscriptions from different sources, we have to make sure concurrent request() invocations are not lost and are routed to the proper Flow.Subscription along with any concurrent cancel() calls. Perhaps the difficulty is lessened by the fact that switching sources happens on a terminal event boundary only, thus we don't have to worry about the old source calling onNext while the logic switches to the new source and complicating the accounting of requested/emitted item counts. Enter SubscriptionArbiter. Subscription arbitration We have to deal with 4 types of potentially concurrent signals when arbitrating Flow.Subscriptions: A request(long) call from downstream that has to be routed to the current Flow.Subscription A cancel() call from downstream that has to be routed to the current Flow.Subscription and cancel any future Flow.Subscription. A setSubscription(Flow.Subscription) that is called by the current Flow.Subscriber after subscribing to any Flow.Publisher which is not guaranteed to happen on the same thread subscribe() is called (i.e., as with the standard SubmissionPublisher or our range() operator). A setProduced(long n) ## Java 9 Flow API: timing out events DevFeed: [Java 9 Flow API: timing out events]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-25T11:46:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Java 9](), [Concurrency](), [Programming](), [Streams]() Tags: [await](), [concurrency](), [flow](), [flow-api](), [java-9](), [jdk](), [programming](), [reactive](), [reactive-programming](), [streams](), [subscriber](), [subscription](), [timeout]() ### AI overview This tutorial explains how to design a timeout operator for Java 9 Flow API publishers. It models each event as a race between item arrival and a timer, signals a TimeoutException when the timer wins, and addresses serialized downstream signals when timeout and upstream events occur concurrently. ### Source excerpt Introduction One of the main properties of reactive programming is that the events may arrive over time instead of immediately available to a consumer. In traditional Future-based programming, one could wait for the result in a blocking manner via Future.get(long, TimeUnit). Other data sources, such as network InputStream have either their own built-in timeout facility or one has to use external means to close the stream after certain period of time to unblock the reader to it. Java 8 Streams have also no direct timeout support. In the reactive mindset, one can consider timing out events (items) as requesting an element and racing its arrival against the clock. If the item arrives in time, we should ignore the clock. If the clock fires first, we should stop the sender of the items and somehow notify the consumer of the situation. Perhaps the simplest way is to signal onError with a TimeoutException. Since there could be multiple items from a flow, we have to do this racing for each potential items over and over until the flow terminates. The timeout operator Since there is "time" in timeout, we'll need a source of time that can be started and stopped at will. The first tool that comes into mind is the java.util.Timer class, however, even its Javadoc suggest one uses a ScheduledExecutorService instead. If one has to deal with a lot of timed operations, besides of timing out flows, having the control over such signals via a (set of) ScheduledExecutorServices is desirable. Therefore, let's define our timeout API with it: public static Flow.Publisher timeout( Flow.Publisher source, long timeout, TimeUnit unit, ScheduledExecutorService timer) { return new TimeoutPublisher<>(source, timeout, unit, timer); } (Note that if one uses the Executors.newScheduledExecutorService(), it has to be shutdown at some point, otherwise it's non-daemon thread by default would prevent the JVM from quitting.) One primary responsibility of this type of operator is to make sure the ## Java 9 Flow API: switching threads DevFeed: [Java 9 Flow API: switching threads]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-21T10:49:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [API](), [reactive](), [Concurrency](), [Concurrent Programming]() Tags: [asynchronous](), [executor](), [executorservice](), [flow](), [flow-api](), [idea](), [intellij](), [java](), [java-9](), [jvm](), [main-thread](), [observeon](), [publisher](), [thread](), [threading](), [threads]() ### AI overview This tutorial explains how to switch execution between threads in reactive flows using Java 9's Flow API. It compares embedding thread switching in operators with a separate observeOn stage, then outlines an observeOn implementation based on Executor, asynchronous boundaries, bounded queues, and request management. ### Source excerpt Introduction Ensuring certain computations happen on the right thread, usually off the main thread, is a very common development task when dealing with reactive flows. When building up tools for Java 9's Flow API, one can decide to add this thread-switching support to each operator directly - see the range() operator from the start of the series -, or have a standalone stage for this purpose. This is a tradeoff. Inlining thread switching avoids bogging down the source thread like the thread-stealing behavior of most of the queue-drain approach presented so far. A separate operator allows better composition and may even allow working with exotic asynchrony-providing components. The observeOn operator In Java, threading support is provided via the Executor, ExecutorService and ScheduledExecutorService-based API. Executor is is the most basic one of them which only provides a single execute(Runnable) method. This allows creating an Executor from a lambda: Executor trampoline = Runnable::run; Executor swing = SwingUtilities::invokeLater; Executor pool = ForkJoinPool.commonPool(); As the least common denominator, we'll use Executor in defining our observeOn operator: public static Flow.Publisher observeOn( Flow.Publisher source, Executor exec, int prefetch) { return new ObserveOnPublisher<>(source, exec, prefetch); } Crossing an asynchronous boundary requires the temporary storage of an event until the other side can pick it up. The queue-drain approach can provide a nice bounded queue we can size with prefetch. In addition, the so-called stable-prefetch request management (shown in the mapFilter operator before) allows minimizing the overhead of requesting more items. First, let's see the skeleton of the operator's main Flow.Subscriber implementation: static final class ObserveOnSubscriber implements Flow.Subscriber, Flow.Subscription, Runnable { final Flow.Subscriber downstream; final Executor exec; final int prefetch; final Queue queue; F ## Interoperation between RxJava and Kotlin Coroutines DevFeed: [Interoperation between RxJava and Kotlin Coroutines]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-11T21:29:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [kotlin-coroutines](), [Coroutines](), [Kotlin](), [Code](), [Compiler]() Tags: [asynchronous](), [await](), [backpressure](), [channel](), [code](), [compiler](), [coroutines](), [flow](), [flowable](), [flowablesubscriber](), [interoperation](), [kotlin](), [kotlin-coroutines](), [notify](), [rxjava](), [stream](), [subscription](), [suspend]() ### AI overview This tutorial explains how to make RxJava work with Kotlin Coroutines. It introduces a suspendable emitter and a coroutine-based Flowable generator that can suspend emission until downstream demand is available. ### Source excerpt Introduction Writing imperative-looking code with Kotlin Coroutines is certainly an attractive property of it, but I'd think things can get quite convoluted pretty fast once, for example, Selectors are involved. I haven't gotten there to look at what Selectors are, I only read that they can help you implement a flatMap like stream combiner. We are not goind to do that now, RxJava can do it for us after all. However, the reasonable question arises: if I have a coroutine generator, a coroutine transformation or simply want to receive items from a Flowable, how can I make RxJava work with these coroutines? Easily with the combined magic of Kotlin Coroutines and RxJava coroutines! Suspendable Emitter A generator is a source-like construct that emits items followed by a terminal signal. It should be familiar from RxJava as the Flowable.generate() operator. It gives you a FlowableEmitter and the usual onNext, onError and onComplete calls on it. One limitation is that you can call onNext only once per invocation of your (Bi)Consumer lambda that receives the emitter. The reason is that we can't block a second call to onNext and we don't want to buffer it either; therefore, RxJava cooperates with the developer. Compiler supported suspension and state machine built by it, however, allow us to prevent a second call from getting through by suspending it until there is a demand from the downstream, which then resumes the coroutine where it left off. Therefore, we can lift the single onNext requirement for our Coroutine-based generator. So let's define the SuspendEmitter interface interface SuspendEmitter : CoroutineScope { suspend fun onNext(t: T) suspend fun onError(t: Throwable) suspend fun onComplete() } By extending the CoroutineScope, we provide useful infrastructure (i.e., coroutineContext, isActive) to the block that will target our SuspendEmitter. One can argue that why use onError and onComplete since a coroutine can throw and simply end. The reason is that this w ## Java 9 Flow API: ordered merge DevFeed: [Java 9 Flow API: ordered merge]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-11T14:46:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [API]() Tags: [backpressure](), [batching](), [flow](), [flow-api](), [java](), [java-9](), [merge](), [ordered-merge](), [queue](), [reactive](), [rxjava](), [sequences](), [stream]() ### AI overview This article explains why zip() and flatMap() are unsuitable for merging multiple ordered event sequences while preserving order. It presents an orderedMerge() operator that selects the smallest or largest available item, requires a fixed number of source sequences, and discusses how unordered sources produce priority-queue-like output. It also introduces prefetching, queues, coordination, batching, and stable-prefetch backpressure for an inner consumer implementation. ### Source excerpt Introduction Sometimes, one has several ordered sequences of events and would like to merge them into one single flow. Since one element from a sequence should come before another element in another sequence, we need a way to keep comparing elements with each other from different sequences. Unfortunately, zip() doesn't work because it takes a row of available items and item #2 from sequence #2 may come before item #1 from stream #3. Plus, if one stream is shorter than the others, the end sequence stops. Similarly, flatMap() doesn't work because it takes the next item from any inner source sequence the moment it is available without any ordering considerations at that point. At least it emits all items from all sources (provided there are no errors of course). Therefore, we need something between the two operators: one that collects up a row of items from the sources, decides which is the smallest/largest of them based on some comparison logic and only emits that. It then awaits a fresh item from that specific source (or completion) and repeats the picking of the smallest/largest item as long as there are requests for it. Such operator, let's call it orderedMerge(), has an implication about the number of its inner source sequences: it has to be fixed. The reason for it is that it has to pick the smallest/largest of the available items in order for the output to be in order. If there is still a source missing, it can't know for sure the others are smaller/larger that any of the upcoming item from that missing source will produce. The second implication is, what happens if the sources themselves are not ordered? The logic presented in this post still works, but the end output won't be totally ordered. It will act like some priority queue instead: picking important items first before turning to less important ones. The inner consumer Operators handling multiple sources often need a way to prefetch item from these sources and give out them on demand to some joining logic ## Rewriting RxJava with Kotlin Coroutines? DevFeed: [Rewriting RxJava with Kotlin Coroutines?]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-09T15:16:00Z Content type: opinion Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Coroutines](), [Kotlin](), [kotlin-coroutines](), [reactive](), [Streams]() Tags: [callback](), [cancellation](), [coroutine](), [coroutines](), [kotlin](), [kotlin-coroutines](), [library](), [reactive](), [reactive-streams](), [rxjava](), [scrabble]() ### AI overview The article explores whether a declarative-reactive library can be implemented with Kotlin coroutines as an alternative to RxJava. It examines coroutine abstractions, operator design, cancellation, lazy versus eager execution, and the potential trade-offs for library developers and users. ### Source excerpt Introduction Someone influential stated that RxJava should be rewritten with Kotlin Coroutines. I haven't seen any attempt of it as of now and declaring such a thing to be (not) worth without actually trying is irresponsive. As we saw in the earlier post and the response in the comment section, following up on the imperative-reactive promise leads to some boilerplate and questionable cancellation management, and the idiomatic Kotlin/Coroutine enhancement suggested is to ... factor out the imperative control structures into common routines and have the user specify lambda callback(s); thus it can become declarative-reactive, just like RxJava interpreted from a higher level viewpoint. Kind of defeats one of the premises in my understanding. This doesn't diminish the power of coroutine-based abstraction but certainly implies a relevant question: who is supposed to write these abstract operators? One possible answer is, of course, library writers who not only have experience with abstracting away control structures but perhaps wield deeper knowledge about how the coroutine infrastructure can be utilized in certain complicated situations. If this assumption of mine is true, that somewhat defeats another premise of coroutines: the end user will likely have to stick to writing suspendable functionals and discover operators provided by a library most of the time. So what's mainly left is to see if implementing a declarative-reactive library on top of coroutines gives benefits to the library developer (i.e., ease of writing) over hand crafted state-machines and (reasonable) performance to the user of the library itself. The library implementation Perhaps one of the more attractive properties of RxJava is the deferred lazy execution of a reactive flow (cold). One sets up a template of transformations and issues a subscribe() call to begin execution. In contrast, CompletableFuture and imperative Coroutines can be thought as eager executions - in order to retry them one has to ## Java 9 Flow API: mapping asynchronously DevFeed: [Java 9 Flow API: mapping asynchronously]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-07T14:00:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [reactive](), [Concurrency](), [implementation](), [API]() Tags: [api](), [asynchronous](), [concatmap](), [flow](), [flow-api](), [java](), [java-9](), [jdk](), [map](), [mapping](), [reactive](), [rxjava](), [varhandle]() ### AI overview This tutorial presents the mapWhen operator for asynchronously mapping Java 9 Flow.Publisher values. It explains how the operator supports one-at-a-time mapping, limits inner publishers to one element, and can combine the original and mapped values. ### Source excerpt Introduction There are cases where mapping an upstream value of type T has to be mapped to type U , one-for-one, but the mapping process itself involves asynchronous work. With RxJava, this is a de-facto use case for concatMap, concatMapEager and flatMap, depending on the concurrency expectations about the mapping itself (i.e., one at a time, multiple at once but in-order and arbitrary order respectively). Let's assume we don't want to run multiple concurrent mapping thus concatMap would suffice. We can (and will in a future post) write that operator, but we should face two additional challenges: the standard Java 9 Flow API has no notion of 0..1 reactive type so we have to restrict the inner Flow.Publisher to at most one element (take(1)); and we'd sometimes zip the original and the mapped result into a third type R. These requirements warrant their own custom operator, enter mapWhen(). The mapWhen operator I must admit, the name comes from Reactor-Core after they picked my implementation named mapAsync() from RxJava 2 Extensions. It certainly matches the naming of other operators, such retryWhen(), but arguably the function parameter signature is different (i.e., not a Publisher -> Publisher transformation): public static Flow.Publisher mapWhen(Flow.Publisher source, Function> mapper) { return mapWhen(source, mapper, (t, u) -> u); } public static Flow.Publisher mapWhen(Flow.Publisher source, Function> mapper, BiFunction combiner ) { return new FlowMapWhen<>(source, mapper, combiner); } One would think that supporting the combiner case with the same operator implementation adds unreasonable overhead. We'll see later that this is not the case because both the original and mapped value will be available in a way that makes application (t, u) -> u bi-function a trivial, and when JIT-ed, a fall-through case. I'll omit the outer Fl ## RxJava vs. Kotlin Coroutines, a quick look DevFeed: [RxJava vs. Kotlin Coroutines, a quick look]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-05T15:11:00Z Content type: comparison Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [kotlin-coroutines](), [Kotlin](), [Coroutines](), [reactive](), [Usability](), [Programming]() Tags: [asynchronous](), [blocking](), [business-logic](), [code](), [coroutine](), [coroutines](), [experimental](), [kotlin](), [kotlin-coroutines](), [main-thread](), [programming](), [reactive](), [reactive-programming](), [retry](), [runblocking](), [rxjava](), [suspend](), [thread](), [timeout](), [zip]() ### AI overview This article compares RxJava and Kotlin Coroutines through an example involving delayed unreliable services, timeouts, cancellation, retries, and combining results. It emphasizes usability over raw performance and describes coroutine code as more sequential and synchronous-looking, while noting that coroutines were experimental at the time. ### Source excerpt Introduction Does Kotlin Coroutines make RxJava and reactive programming obsolete? The answer depends on who you ask. Enthusiasts and marketing departments would say yes without hesitation. If so, sooner or later developers would have to convert Rx code into coroutines or write something with coroutines from the start. Since Coroutines are currently experimental, there is always the prospect deficiencies, especially regarding the overhead, will be resolved eventually. Therefore, this post will focus more on usability than raw performance. The scenario Let's say we have two functions imitating unreliable service: f1 and f2, both returning a number after some delay. We have to call these services, sum up their returned values and present it to the user. However, if this doesn't happen within 500 milliseconds, we don't expect it to happen reasonably faster, thus we'd like to cancel and retry the two services for a limited amount of time before giving up after some number of retries. The Coroutine Way Programming via coroutines feels like programming with the traditional ExecutorService- and Future-based toolset with the difference that the underlying infrastructure will use suspension, state machine(s) and task rescheduling instead of blocking a thread. First, we need the functions that exhibit the delaying behavior: suspend fun f1(i: Int) { Thread.sleep(if (i != 2) 2000L else 200L) return 1; } suspend fun f2(i: Int) { Thread.sleep(if (i != 2) 2000L else 200L) return 2; } Functions that participate in a coroutine execution should be declared with the suspend keyword and executed within a coroutine context. For demonstration purposes, the logic will sleep for 2 seconds if the parameter supplied to the functions is not 2. This will give a chance to the timeout logic to kick in yet the 3rd attempt to succeed before the timeout. Since going asynchronous usually ends up leaving the main thread, we need a way to block it until the business logic completes before letting the ## Java 9 Flow API: mapping and filtering in one stage DevFeed: [Java 9 Flow API: mapping and filtering in one stage]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-02T14:33:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java 9](), [reactive](), [API](), [Library](), [Open Source]() Tags: [backpressure](), [exception](), [flow](), [flow-api](), [java-9](), [publisher](), [reactive](), [reactive-streams](), [rxjava](), [subscriber]() ### AI overview This article examines how to combine mapping and filtering into a single custom operator for Java 9 Flow API publishers. It discusses API design choices, handling mapped, dropped, failed, and completed items, and preserving Reactive Streams backpressure and protocol requirements. ### Source excerpt Introduction In most reactive libraries, mapping and filtering can be done on a flow via separate operators map() and filter() respectively. One rare occasions, the functions to these operators would need to communicate with each other without sharing information in a flow-external manner and without using defer(). Such combined and standard mapFilter() operator doesn't exist and one has to write one of its own. Given that the Java 9 Flow API is brand new, one has to definitely write a custom operator for it as Java 9 itself doesn't provide any rich set of predefined operations on Flow.Publishers unlike its dual, the Stream API. Shameless advertising By the way, if you are looking for a Java 9 Flow-based, native and modern reactive library with rich set of operators, similar to RxJava 2 (even including some operators from its extension project), I happen to have one for you: Reactive4JavaFlow. It is free and open-source with the promising outlook that one day, it may form the basis for the next major RxJava version... MapFilter API design When the Reactive4Java library was first concieved in 2011, the first significant stumbling block was not the lack of lambdas in Java 6/7 but the lack of extension methods. C# had it and made Rx.NET conveniently extendable (assuming you managed to understand how to write operators for it as it wasn't open source at the time). Java still doesn't have any sign of ever getting extension methods, therefore, we either need a rich abstract base class, such as Flowable or Flux, or an utility class whose methods almost look like extension method definitions with the exception that the developer has to stack them on top of one another: import java.util.concurrent.*; import static FlowUtils.*; Flow.Publisher f = timeout( mapFilter( new FlowRange(1, 10, Runnable::run), (v, e) -> { if (v % 2 == 0) { e.next(v.toString()) } if (v == 7) { e.complete(); } } ), 5, TimeUnit.MILLISECONDS ); When thinking about a combined map and filter operat ## Java 9 Flow API: asynchronous integer range source DevFeed: [Java 9 Flow API: asynchronous integer range source]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-03-05T13:08:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [reactive](), [interfaces](), [IntelliJ IDEA]() Tags: [asynchronous](), [backpressure](), [coroutine](), [flow](), [flow-api](), [java](), [java-9](), [jdk](), [operator](), [publisher](), [range](), [reactive](), [reactive-streams](), [request-management](), [source](), [submissionpublisher](), [subscriber](), [subscription](), [trampoline](), [varhandle]() ### AI overview A tutorial explores Java 9's Flow API and Reactive Streams interfaces by building an asynchronous integer range Publisher. It discusses composing Publishers, implementing Flow.Subscription, handling subscriber demand, and using IntelliJ 2017.1 EAP while the APIs were non-final. ### Source excerpt Introduction Java 9 is becoming more reactive by introducing the Reactive-Streams interfaces under the parent class java.util.concurrent.Flow, enabling a new standard interoperation between future libraries built on top. There is almost no documentation beyond a underwhelming Oracle documentation and the SubmissionPublisher class' JavaDoc about how to write Publishers, Subscriptions and Subscribers under the Flow API. Plus the Oracle document practically concludes with see RxJava. Indeed, replacing the imports of org.reactivestreams.* with java.util.concurrent.Flow.* in RxJava 2's sources get's one a fully fledged reactive library but there seems to be one crucial expectation with components built on the Flow API: they have to be asynchronous at every stage. I could argue that the underlying concepts work totally fine in synchronous mode, but who am I to question the established definitions? Oh well, if the constraint is to be asynchronous, then let's do it in an asynchronous way. To see what it takes, we could start with a relatively simple source: an asynchronous integer range. Since both Java 9 and the IDE support is in non-final state, I recommend IntelliJ 2017.1 EAP for this "exercise". Asynchronous integer range source Unfortunately, Java 9 won't introduce any standard fluent API entry point with all the well loved map(), filter(), flatMap() etc. operators but one has to build individual Publishers and compose them stage-by-stage. This involves creating a parent Publisher class with the following typical pattern to host the input parameters of the flow to be observed: import java.util.concurrent.*; public final class FlowRange implements Flow.Publisher { final int start; final int end; final Executor executor; public FlowRange(int start, int count, Executor executor) { this.start = start; this.end = start + count; this.executor = executor; } @Override public void subscribe(Flow.Subscriber subscriber) { // TODO implement } } For brevit ## The Reactive Scrabble benchmarks DevFeed: [The Reactive Scrabble benchmarks]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-12-27T19:01:00Z Content type: article Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Benchmark](), [data-processing](), [Java](), [Code]() Tags: [benchmarks](), [data-processing](), [java](), [performance](), [reactive](), [scrabble](), [streams]() ### AI overview This article explains the Shakespeare Plays (Reactive) Scrabble benchmark, including its purpose, computation, history, and sequential and parallel implementations. It discusses performance differences between RxJava and Java 8 Streams and describes efforts to improve RxJava 2 performance. ### Source excerpt Introduction In the past year, I've been posting benchmark results under the mysterious Shakespeare Plays (Reactive) Scrabble name. In this blog post, I'll explain what this benchmark is, where does it come from, how it works, what the intent is and how to apply it to your favorite and not-yet-benchmarked library. History The benchmark was designed and developed by Jose Paumard and results presented in his 2015 Devoxx talk (a bit long but worth watching). The benchmark measures how fast a certain data-processing library can find the most valuable word from a set of words taken from (one of) Shakespeare's work based on the rules and point schema of Scrabble. RxJava at the time was in its 1.0.x version and to my surprise, it performed poorly compared to Java 8 Streams: https://youtu.be/fabN6HNZ2qY?t=8369 The benchmark, utilizing JMH, is completely synchronous; no thread hopping happens yet RxJava performs 10x slower, or more likely, it has 10x more overhead in the associated set of operators. In addition, Jose also added a parallel-stream version which runs the main "loop" in parallel before joining for the final result. More disappointingly, RxJava 2 developer preview the time was terrible as well (relatively, measured on a weak CPU in February). Therefore, instead of blaming the benchmark or the author, I set out on a quest to understand the benchmark's expectations and improve RxJava 2's performance and if possible, port that back to RxJava 1. The original Stream-benchmark Perhaps the most easy way to understand how the computation in the benchmark works, Let's see the original, non-parallel Stream version of it. Since going sequential or parallel requires only a sequential() or parallel() operator on a Stream, they both extend an abstract superclass containing the majority of the code and only get specialized for the operation mode in two additional classes. ShakespearePlaysScrabbleWithStreamBeta.java I added postfix "Beta" - meaning alternate version in this cont ## Async Iterable/Enumerable vs. Reactive-Streams DevFeed: [Async Iterable/Enumerable vs. Reactive-Streams]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-05-02T13:54:00Z Content type: comparison Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Java](), [Promise](), [interfaces](), [Library](), [implementation]() Tags: [async](), [backpressure](), [interface](), [java](), [library](), [reactive-streams]() ### AI overview This article compares Java Async Iterables, also called Async Enumerables in C#, with RxJava and Reactive Streams. It explains how asynchronous MoveNext operations provide backpressure and describes a Java 8 implementation using IAsyncEnumerable, IAsyncEnumerator, CompletionStage, and cancellation support. ### Source excerpt Introduction Backpressure is essential if one wants to avoid buffer bloat and excessive memory usage if two stages in a reactive pipeline consume events with different speed. RxJava and Reactive-Streams developed a non-blocking, request-coordinating protocol to solve this problem, but you may have heard there are alternatives to it. One alternative that comes up from time to time is Async Iterables (Java terminology) or Async Enumerables (C# terminology). In fact Rx.NET has an Ix.NET (stands for Interactive Extensions) sub-project in which there is the Async Enumerables library. It solves this backpressure problem by having a Task (~ CompletableFuture, ~ Promise) returned from its MoveNext() (~ hasNext()) method and when that Task fires, you can consume the Current property (~ next() method). The backpressure behavior comes from the fact that you'd call MoveNext() again only after you processed the the current element. Unfortunately, I haven't found a Java implementation for the IAsyncEnumerable (haven't really looked beyond a few Google searches), so I decided I'll implement it on my own in Java 8, see what it takes to get data across with it and how performant is it compared to my current cutting-edge understanding of reactive-flows: the Reactive-Streams-Commons library. Base API Since Async Enumerables are designed in deferred execution in mind, the base API consists of two interfaces: interface IAsyncEnumerable { IAsyncEnumerator enumerator(); } interface IAsyncEnumerator { CompletionStage moveNext(CompositeSubscription cancel); T current(); } The IAsyncEnumerable is the equivalent of Iterable and it hands out IAsyncEnumerators. IAsyncEnumerator has a moveNext method which returns a CompletionStage indicating if there is value available via current() (signals true) or the sequence ended (signals false). C# CancellationToken looks like our CompositeSubscription so I'm reusing it as the way for cancellation. (Sidenote: I'm not sure how cancellati ## Google Agera vs. ReactiveX DevFeed: [Google Agera vs. ReactiveX]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-04-24T11:44:00Z Content type: comparison Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Android](), [android-development](), [API](), [Programming]() Tags: [android](), [android-development](), [api](), [apis](), [java](), [reactive-programming](), [rxjava]() ### AI overview This article compares Google's Agera reactive library for Android with established reactive libraries including RxJava, Reactor, and Akka-Streams. It describes Agera's valueless Observer and Updatable APIs, then examines subscription and pipeline contention issues in those designs and related reactive APIs. ### Source excerpt Introduction If you are following events around Android development, or just happen to follow all things reactive, there was a "big" announcement from Google: they've released their reactive programming library targeting Android specifically: Agera. Of course, one has to look into the details to get an accurate picture. "By Google" means a team in Google working on Google Play Movies. Certainly its sounds more amplified to say Google than the full path to the team. I happen to do this as well when someone asks where I work: in a lab at the Hungarian Academy of Sciences instead of at the Engineering and Management Intelligence Research Laboratory at the Institute for Computer Science and Control of the Hungarian Academy of Sciences. (Plus, you don't get tired and lost while I'm emitting these words :) It doesn't really matter who released it, all that matters what they released and how it relates to the well established reactive libraries, RxJava, Reactor and Akka-Streams, altogether. The Core API The Agera library is built around the valueless Observer pattern: Observables take Updatables and signal change via update() calls. It is then the responsibility of those Updatables to figure out what changed. This is practically a zero argument reactive dataflow which relies on side-effects per update(). interface Updatable { void update(); } interface Observable { void addUpdatable(Updatable u); void removeUpdatable(Updatable u); } They look innocent and reactive, right? Unfortunately, they've run into the issue with the original java.util.Observable and the other addListener/removeListener based reactive APIs (which I categorized as 0th generation). Agera Observable The problem with this pair of methods is that every Observable who adds behavior over an incoming Updatable has to remember the original Updatable in some whay for the case when the same Updatable is removed: public final class DoOnUpdate implements Observable { final Observable source; final Runnable action; ## Operator fusion (part 2 - final) DevFeed: [Operator fusion (part 2 - final)]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-04-19T14:57:00Z Content type: article Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Protocol (disambiguation)](), [Streams](), [Java]() Tags: [java](), [protocol](), [reactive](), [reactive-streams](), [rxjava](), [streams]() ### AI overview This article explains the API and protocol mechanisms behind operator fusion in Reactive Streams. It focuses on macro-fusion for synchronous sources that emit zero or one element, including just(), empty(), and fromCallable(), and describes using Callable and ScalarCallable to enable optimization. ### Source excerpt Introduction In the previous part, I've introduced the concepts around operator fusion. In this post, I'll detail the API and protocols required make operator fusion happen. In its current form, operator fusion works between two subsequent operators and is based on the ability to identify each other and, in case of micro-fusion, switch to a different protocol than Reactive-Streams (RS) if both agree. Macro-fusion constructs The primary targets of macro-fusion are the single element sources: just(), empty(), fromCallable(). Firing up the complete RS infrastructure for such single elements is quite expensive, but half of the API use in RxJava and Reactor come from these. Therefore, RxJava introduced Single and Reactor introduced Mono to help as much as possible and offer (ever increasingly) optimized operators on them. However, knowing a source will generate 0 or 1 element during assembly time is also a great help in regular Observable / Flux uses. In addition, knowing the source is also a constant helps inlining it in via some custom operator. Creating 0 or 1 element synchronous sources To indicate a source returns a single value, the Reactive-Streams-Commons (Rsc) project (and Reactor off it) established a contract: If a Publisher implements java.util.concurrent.Callable, it is considered a 0 or 1 element source. You can implement Callable and return a non-null value that can be computed synchronously. You can also return null which indicates an empty result. (Remember, RS doesn't allow null values over onNext.) The call to call() will happen during subscription time. public class MySingleSource implements Publisher, Callable { @Override public void subscribe(Subscriber s) { s.onSubscribe(new ScalarSubscription<>(s, System.currentTimeMillis())); } @Override public Object call() throws Exception { return System.currentTimeMillis(); } } If the 0 or 1 element source is known to be constant, the source can be the subject of assembly time ## SubscribeOn and ObserveOn DevFeed: [SubscribeOn and ObserveOn]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-31T12:55:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Programming](), [reactive](), [User Experience](), [Android](), [GUI]() Tags: [backpressure](), [blocking](), [cancellation](), [executorservice](), [main-thread](), [observeon](), [reactive](), [rxjava](), [subscribeon](), [thread]() ### AI overview A tutorial explaining why RxJava's subscribeOn and observeOn operators are often confused. It distinguishes their effects by tracing subscription and method-call flow, and discusses moving subscription side effects such as network, database, or blocking work away from the current thread. ### Source excerpt Introduction One of the most confused operator pair of the reactive ecosystem is the subscribeOn and observeOn operators. The source of confusion may be rooted in a few causes: they sound alike, they sometimes show similar behavior when looked at from downstream and they are duals in some sense. It appears the name-confusion isn't local to RxJava. Project Reactor faces a similar issue with their publishOn and dispatchOn operators. Apparently, it doesn't matter what they are called and people will confuse them anyhow. When I started learning about Rx.NET back in 2010, I never experienced this confusion; subscribeOn affects subscribe() and observeOn affects onXXX(). (Remark: I've searched Channel 9 for the early videos but couldn't really find the talk where they build up these operators just like I'm about to do. The closest thing was this.) My "thesis" is that the confusion may be resolved by walking through how one can implement these operators and thus showing the internal method-call flow. SubscribeOn The purpose of subscribeOn() is to make sure side-effects from calling subscribe() happens on some other thread. However, almost no standard RxJava source does side-effects on its own; you can have side-effects with custom Observables, wrapped subscription-actions via create() or as of lately, the with the SyncOnSubscribe and fromCallable() APIs. Why would one move the side-effects? The main use cases are doing network calls or database access on the current thread or anything that involves blocking wait. Holding off a Tomcat worker thread hasn't been much of a programming problem (that doesn't mean we can't improve the stack with reactive) but holding off the Event Dispatch Thread in a Swing application or the Main thread in an Android application has adverse effect on the user experience. (Sidenote: it's a funny thing that blocking the EDT is basically a convenience backpressure strategy in the GUI world to prevent the user from changing the application state whil ## Writing a custom reactive base type DevFeed: [Writing a custom reactive base type]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-20T13:18:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Java](), [Programming](), [Code]() Tags: [code](), [java](), [programming](), [reactive](), [rxjava]() ### AI overview This tutorial explains how to create a custom reactive base type around RxJava's Observable. It shows how wrapping can add custom operators, hide unsuitable operators, and provide a distinct type for specialized processing pipelines, while also discussing interoperability with other observable types. ### Source excerpt Introduction From time to time, the question or request comes up that one would really like to have his/her own reactive type. Even though RxJava's Observable has plenty of methods and extension points via lift(), extend() and compose(), one feels the Observable should have the operator xyz() or in some chains, the chain shouldn't allow calling uvw(). The first case, namely adding a new custom method without going through the project as a contribution, is as old as the reactive programming on the JVM. When I first ported Rx.NET to Java, I had to face the same problem because .NET had the very convenient extension method support already back in 2010. Java doesn't have this and the idea has been rejected in the version 8 development era in the favor of default methods with the "justification" that such extension methods can't be overridden. Yes they can't but they can be replaced by another method from another class. The second case, hiding or removing operators, comes up with custom Observables where certain operations don't make sense. For example, given a ParallelObservable that splits the input sequence into parallel processing pipelines internally, it makes sense to map() or filter() in parallel, but it doesn't make sense to use take() or skip(). Wrapping Both cases can be solved by writing a custom type and just wrap the Observable into it. public final class MyObservable { private Observable actual; public MyObservable(Observable actual) { this.actual = actual; } } Now we can add operators of our liking: // ... public static MyObservable create(Observable o) { return new MyObservable(o); } public static MyObservable just(T value) { return create(Observable.just(value)); } public final MyObservable goAsync() { return create(actual.subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread())); } public final MyObservable map(Func1 mapper) { return create(actual.map(mapper)); } public final void subscribe( ## Operator-fusion (Part 1) DevFeed: [Operator-fusion (Part 1)]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-11T13:06:00Z Content type: article Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [Programming]() Tags: [implementation](), [programming](), [reactive](), [reactive-programming](), [reactive-streams](), [rxjava]() ### AI overview This article introduces operator fusion in reactive programming: combining successive operators to reduce dataflow overhead in time and memory. It discusses optimization techniques, experimentation in the reactive-streams-commons repository, and relationships to RxJava, Project Reactor, and Akka Streams. ### Source excerpt Introduction Operator-fusion, one of the cutting-edge research topics in the reactive programming world, is the aim to have two of more subsequent operators combined in a way that reduces overhead (time, memory) of the dataflow. (Other cutting-edge topics are: 1) reactive IO, 2) more native parallel async sequences and 3) transparent remote queries.) The key insight with operator-fusion is threefold: many sequences are started from constant or quasi-constant sources such as just(), from(T[]), from(Iterable), fromCallable() which don't really need the thread-safety dance in a sequence of operators, some pairs of operators can share internal components such as Queues and some operators can tell if they consumed the value or dropped it, avoiding request(1) call overhead. In this mini-series, I'll describe the hows and whys of operator-fusion, as we currently understand it. By "we", I mean the joint research effort on optimizing Reactive-Streams operators beyond what's there in RxJava 2.x and has been in previous versions of Project Reactor. The experimentation happens in the reactive-streams-commons, Rsc for short, GitHub repository. The results of the Rsc is now driving Project Reactor 2.5 (currently in milestone 2) and verified by a large user base. Hopefully, RxJava can benefit from the results as well (but maybe not before 3.x). If you are following Akka-Streams, you might have read/head about operator-fusion there as well. As far as I could understand their approach, their objective is to make sure more stages of the pipeline run on the same Actor, avoiding the previous, very likely, thread-hopping with their sequences. Essentially, there is now a mode where the developer can define the async boundaries in the pipeline. Does this sound familiar? From day 1, Rx-based libraries let you do this. Generations Reactive libraries and associated concepts evolved over time. What we had 7 years ago in Rx.NET, requirements and implementation-wise is significantly different w ## RxJava design retrospect DevFeed: [RxJava design retrospect]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-05T19:32:00Z Content type: opinion Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [reactive](), [interfaces](), [implementation](), [Code]() Tags: [blog-post](), [code](), [implementation](), [interfaces](), [rxjava]() ### AI overview This retrospective examines design and implementation decisions in RxJava, focusing on synchronous cancellation. It explains how RxJava's early observable and observer interfaces could prevent timely disposal of a running sequence, compares this with Rx.NET's scheduling approach, and discusses performance differences in a range operator benchmark. ### Source excerpt Intoduction RxJava is now out more than 3 years and lived through several significant version changes. In this blog post, I'll point out design and implementation decisions that I personally think wasn't such a good idea. Don't get me wrong, it doesn't mean that RxJava is bad or I knew all along how to do it "properly". It was a learning process for all of us involved, but the question is, can we learn from those mistakes and do it better in the next major version? Synchronous unsubscription In the early days, RxJava mirrored the architecture of Rx.NET which consisted of two important interfaces, IObservable and IObserver, derived through dualizing the IEnumerable and IEnumerator. (This was also true for my own library, Reactive4Java). If we look at IObservable, we find the subscribe() method that returns an IDisposable. This returned object allows one to dispose or cancel a running sequence. However, it has a critical problem I demonstrate with a minimalistic reactive program: interface IDisposable { void dispose(); } interface IObserver { void onNext(T t); } interface IObservable { IDisposable subscribe(IObserver observer); } IObservable source = o -> { for (int i = 0; i < Integer.MAX_VALUE; i++) { o.onNext(i); } return () -> { }; }; IDisposable d = o.subscribe(System.out::println); d.dispose(); If we run this code, it starts to print a lot of numbers to the console, despite we called dispose on the returned object by the subscribe method. What's wrong? The problem is that the source observable can only return its IDisposable object only after the for-loop finishes, but then it has nothing to do. The whole setup is synchronous and thus this structure can't be reasonably cancelled. Although Rx is good at async processing, many steps in a typical pipeline is synchronous and is affected by this synchronous cancellation requirement. Since Rx.NET is at least 3 years older than RxJava, how could this shortcoming still be in today's Rx.NET? The example ## FlatMap (part 2) DevFeed: [FlatMap (part 2)]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-02T22:50:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Concurrency](), [implementation](), [Network](), [Data structures]() Tags: [backpressure](), [concurrency](), [concurrent](), [errors](), [implementation](), [network](), [parameter](), [performance](), [queue]() ### AI overview This post extends a flatMap implementation with bounded concurrency and delayed error handling. It uses backpressure to limit active inner Observables and collects errors for a final CompositeException. ### Source excerpt Introduction In this post, we will look into expanding the features of our flatMap implementation and improve its performance. RxJava's flatMap implementation offers limiting the maximum concurrency, that is, the maximum number of active subscriptions to the generated sources and allows delaying exceptions coming from any of the sources, including the main. Limiting concurrency Due to historical reasons, RxJava's flatMap (and our version of it from part 1) is unbounded towards the main source. This may work with infrequent main emissions and/or short lived inner Observable sequences. However, even if the main source, such as range(), can emit at any rate, the mapped inner Observables may consume limited resources such as network connections. So the question is, how can we make sure only an user defined number of active Observables are being merged at once? How can we make sure some source emits only a limited number of values? The answer is, of course, backpressure. To limit the concurrency in flatMap, the idea is to request a maxConcurrency amount upfront via request(), and then whenever a source completes, request(1) extra. Let's change our OpFlatMap and FlatMapSubscriber's implementation to include this maxConcurrency parameter: final int maxConcurrency; public OpFlatMap(Func1> mapper, int prefetch, int maxConcurrency) { this.mapper = mapper; this.prefetch = prefetch; this.maxConcurrency = maxConcurrency; } @Override public Subscriber call(Subscriber t) { FlatMapSubscriber parent = new FlatMapSubscriber<>(t, mapper, prefetch, maxConcurrency); parent.init(); return parent; } As a contract, we will handle Integer.MAX_VALUE as an indicator for the original unbounded mode: final int maxConcurrency; public FlatMapSubscriber(Subscriber actual, Func1> mapper, int prefetch, int maxConcurrency) { this.actual = actual; this.mapper = mapper; this.prefetch ## Understanding flatMap and merge in RxJava DevFeed: [Understanding flatMap and merge in RxJava]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-02-24T16:22:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Programming](), [Functional programming]() Tags: [backpressure](), [callback](), [consumer](), [express](), [functional](), [map](), [merge](), [protocol](), [rxjava](), [stream]() ### AI overview This introductory article explains the outer and inner properties of flatMap, including how it transforms values into Observables whose outputs may vary in time, location, and count. It compares flatMap with merge, describing flatMap as an upstream value-to-Observable operation and merge as an operation over an Observable of Observables. It also discusses backpressure, serialization, and their relationship in RxJava 1.x. ### Source excerpt Introduction In this blog post, I begin to explain the outer and inner properties of the most used, misunderstood and at the same time, one of the most complex operator there is: flatMap. FlatMap is most useful because it let's you replace simple values with something that can change the output in terms of time, location and value count. FlatMap is misunderstood because it is introduced late, not enough time is spent demonstrating it and often surrounded with functional programming technoblabble. Finally, it's complex because it has to coordinate backpressure of a single consumer and request from multiple sources, and we usually don't know which of them will respond with actual items. Maybe all of them. FlatMap has a companion operator: merge. Merge lets you flatten a sequence of Observables into a single stream of values while ensuring the contract of the Observer, namely, the requirement of non-concurrent invocation of the onXXX methods and the conformance to the onNext* (onError|onCompleted)? protocol. This is necessary because although the individual Observables you merge do conform the same protocol individually, they get mixed in time, location and numbers when you listen to them all at once. Of course, flatMap has to do the same so why are there two operators? The answer is convenience and usage pattern. FlatMap is an in-sequence operator that reacts to values from the upstream by generating an Observable, through a callback function, that is internally subscribed to, coordinated and serialized in respect to any previous or subsequent Observables generated through the same callback function. Merge, on the other hand works on a two-dimensional sequence: an Observable of Observables. There is no function involved here but the operator has to subscribe all of those inner Observables emitted by the outer Observable. The fun thing is, you can express them with the other: Func1> f = ... source.flatMap(f) == Observable.merge(source.map(f)) Observabl