# Reactive Streams Published articles for Reactive Streams. This is one page of public article previews, not the complete archive. Follow Next page to continue. Summaries are not the original full articles. ## Kotlin Coroutines Compared with Reactor for Sequential and Asynchronous Operations DevFeed: [Kotlin Coroutines Compared with Reactor for Sequential and Asynchronous Operations]() Original publisher: [Read original article]() Published: 2025-10-06T00:00:00Z Content type: comparison Language: en Sources: [Kt. Academy]() Topics: [Coroutines](), [Kotlin](), [kotlin-coroutines](), [Reactive Streams](), [RxJava](), [async/await](), [Android](), [Back end]() Tags: [android](), [async](), [asynchronous](), [await](), [backend](), [compare](), [comparison](), [coroutines](), [kotlin](), [kotlin-coroutines](), [programming](), [programming-languages](), [reactive-streams](), [rxjava](), [simple](), [time](), [workshop-learning-programming]() ### AI overview This article compares Kotlin Coroutines with Reactor for backend and Android development. It argues that coroutines provide more readable and straightforward syntax for sequential and asynchronous operations, while Reactor requires additional operators, concepts, and patterns. ### Source excerpt A comparison of Kotlin Coroutines and Reactor, highlighting the elegance and simplicity of Coroutines over the complexity of Reactor. ## Kotlin Flow to RxJava or Reactor and vice versa DevFeed: [Kotlin Flow to RxJava or Reactor and vice versa]() Original publisher: [Read original article]() Published: 2025-09-22T00:00:00Z Content type: tutorial Language: en Sources: [Kt. Academy]() Topics: [kotlin-flow](), [RxJava](), [Reactive Streams](), [interoperability](), [kotlin-coroutines]() Tags: [how-to](), [interoperability](), [kotlin](), [kotlin-flow](), [reactive-streams](), [rxjava](), [workshop-learning-programming]() ### AI overview A guide to converting Kotlin Flow to RxJava or Reactor streams and converting those streams back to Kotlin Flow. It covers interoperability libraries, extensions, coroutine-based stream creation, and handling single-value or no-value operations. ### Source excerpt A guide on how to convert Kotlin Flow to RxJava or Reactor streams and vice versa. ## Why Kotlin Coroutines simplify asynchronous, non-blocking development DevFeed: [Why Kotlin Coroutines simplify asynchronous, non-blocking development]() Original publisher: [Read original article]() Published: 2024-10-09T00:03:00Z Content type: tutorial Language: en Sources: [Kt. Academy]() Topics: [Coroutines](), [Kotlin](), [kotlin-coroutines](), [multiplatform](), [Android](), [RxJava]() Tags: [asynchronous](), [blocking](), [coroutines](), [kotlin](), [kotlin-coroutines](), [multiplatform](), [reactive-streams](), [rxjava](), [workshop-learning-programming]() ### AI overview This tutorial chapter explains why Kotlin Coroutines are useful for asynchronous programming. It highlights their multiplatform support, lightweight execution, non-blocking imperative and reactive styles, reduced boilerplate, and simpler debugging and maintenance compared with callbacks and reactive streams. ### Source excerpt The explanation of why coroutines stand out and offer us what hasn't been offered by other technologies. ## Reactive streams testing DevFeed: [Reactive streams testing]() Original publisher: [Read original article]() Published: 2023-01-08T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [Reactive Streams](), [Testing](), [API](), [Documentation](), [kotlin-flow](), [RxJava](), [Refactoring](), [callback]() Tags: [api](), [code](), [documentation](), [kotlin](), [kotlin-flow](), [reactive-streams](), [refactoring](), [rxjava](), [testing]() ### AI overview This article explains how to test reactive-stream APIs, using RxJava and Kotlin Flow as examples. It emphasizes testing emitted values and other implicit behavior so tests document public API expectations and support safer migrations and refactoring. ### Source excerpt Introduction We should test the code to guarantee that it does exactly what it is expected to do. Tests not only verify correctness of the program but also set expectations. This becomes especially useful for public API, where tests can be treated as part of documentation that describes not only results but also behavior. Such things help do less painful migrations and refactoring as it becomes clear what behaviors are changing and what we should do about that. ## The Many Different Ways to Fetch Data in jOOQ DevFeed: [The Many Different Ways to Fetch Data in jOOQ]() Original publisher: [Read original article]() Author: lukaseder Published: 2022-05-19T07:29:09Z Content type: tutorial Language: en Sources: [jOOQ]() Topics: [jooq](), [Java](), [SQL]() Tags: [collections](), [collector](), [convenience](), [fetching](), [java](), [jdbc](), [joo-f6a0c60d3d07](), [jooq](), [jooq-in-use](), [optional](), [r2dbc](), [reactive-streams](), [sql](), [stream-api](), [streams]() ### AI overview This tutorial explains several ways to fetch data with jOOQ, including eager result-set fetching, external and internal iteration, single-record retrieval, optional results, and resource-managed fetching for large data sets. ### Source excerpt The jOOQ API is all about convenience, and as such, an important operation (the most important one?) like fetch() must come with convenience, too. The default way to fetch data is this: It fetches the entire result set into memory and closes the underlying JDBC resources eagerly. But what other options do we have? Iterable ... Continue reading The Many Different Ways to Fetch Data in jOOQ -> ## Fast Reactor Tests With Virtual Time DevFeed: [Fast Reactor Tests With Virtual Time]() Original publisher: [Read original article]() Author: piotr-kubowicz Published: 2021-12-27T00:00:00Z Content type: tutorial Language: en Sources: [Backend Development on nexocode]() Topics: [reactive](), [Testing](), [Programming](), [Concurrency](), [Code](), [Kotlin]() Tags: [asynchronous](), [code](), [concurrency](), [jvm](), [kotlin](), [network](), [programming](), [reactive](), [reactive-programming](), [reactive-streams](), [testing](), [timeout](), [unit-test]() ### AI overview This tutorial explains how to use Project Reactor's Virtual Time and StepVerifier to test delayed and asynchronous reactive code without waiting for real time to pass. It also discusses controlling external storage behavior and the limitations of polling-based test approaches. ### Source excerpt Sometimes your code deals with a situation when things happen slowly. Maybe you schedule a background task that runs after some time. Or run a special action when asking for data that takes far too long. Either way, it is a tricky case that needs to be tested well. But what to do if we don't want a test that waits a lot? Project Reactor, a reactive programming library for JVM, handles concurrency in a high-level and declarative fashion. Its test utility, StepVerifier, allows using Virtual Time: 'mock' the clock and advance time in your tests faster than the system clock runs. ## Kotlin flow: Nesting vs Chaining DevFeed: [Kotlin flow: Nesting vs Chaining]() Original publisher: [Read original article]() Published: 2021-07-08T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [kotlin-flow](), [Kotlin](), [Reactive Streams](), [reactive](), [RxJava]() Tags: [kotlin](), [kotlin-flow](), [reactive](), [reactive-streams](), [rxjava]() ### AI overview A tutorial comparing nesting and chaining in Kotlin Flow reactive streams. It explains that chaining can improve readability for simple sequential streams, while nesting is useful for sharing data across streams and managing stream lifecycles, such as conditionally observing a user's location. ### Source excerpt Introduction When it comes to reactive streams it is likely everyone heard about huge 'Rx-chains'. But reactive streams are not only about chaining but also about nesting. Let's find out what are they, what are the differences, and why it matters. We'll use Kotlin Flow throughout the article, but everything can be applied to RxJava as well. First of all, we need to come up with definitions. For that, we'll take a look at some simple streams. ## Android Room Hidden Costs DevFeed: [Android Room Hidden Costs]() Original publisher: [Read original article]() Published: 2020-12-18T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [Databases](), [kotlin-flow](), [SQLite](), [Reactive Streams](), [RxJava](), [Android](), [Android Studio]() Tags: [android](), [android-studio](), [databases](), [kotlin-flow](), [reactive-streams](), [rxjava](), [sqlite]() ### AI overview This article examines the hidden costs of using Android Room to support reactive streams. It explains how Room uses SQLite triggers and an internal invalidation table to detect changes, trigger callbacks, and automatically re-query data for observers. ### Source excerpt Introduction Android Room is an awesome AndroidX library. It is great because it provides a clean way of how to deal with databases without introducing some heavy concepts. Out of the box with Room one gets: simplified code comparing with raw SQLite queries working via DAO interfaces and models instead of Cursors auto-generated "boilerplate" code for queries migration support Android Studio tooling (compile-time verification of queries and highlight) support for reactive streams (RxJava, Kotlin Flow) and more In this article, I'd like to discover some hidden costs Room applies to support reactive streams and how one can try to avoid them. ## Working with Kotlin Coroutines and RxSwift - Russell Wolf DevFeed: [Working with Kotlin Coroutines and RxSwift - Russell Wolf]() Original publisher: [Read original article]() Published: 2020-06-15T22:52:17Z Content type: tutorial Language: en Sources: [Touchlab | Enterprise Mobile Innovation & Development]() Topics: [Coroutines](), [Kotlin](), [kotlin-coroutines](), [Reactive Streams](), [iOS](), [Swift](), [RxJava]() Tags: [asynchronous](), [coroutines](), [interop](), [ios](), [kotlin](), [kotlin-coroutines](), [kotlin-multiplatform](), [platforms](), [reactive-streams](), [rxjava](), [swift](), [thread]() ### AI overview A tutorial on type-safe and thread-safe interoperation between Kotlin coroutines in shared code and RxSwift on iOS. It discusses platform limitations, Swift visibility issues, asynchronous event streams, and patterns for handling single- and multiple-event functions, errors, and cancellation. ### Source excerpt We discuss type-safe and thread-safe interop between Kotlin coroutines in shared code, and RxSwift on the iOS side. Useful RxSwift patterns for interop code ## Working with Kotlin Coroutines and RxSwift DevFeed: [Working with Kotlin Coroutines and RxSwift]() Original publisher: [Read original article]() Author: Russell Wolf Published: 2020-06-15T19:07:18Z Content type: tutorial Language: en Sources: [DEV Community 👩💻👨💻: Russell Wolf]() Topics: [kotlin-coroutines](), [Coroutines](), [Kotlin](), [Swift](), [iOS](), [reactive](), [Code]() Tags: [asynchronous](), [coding](), [community](), [coroutines](), [development](), [engineering](), [flow](), [inclusive](), [interop](), [ios](), [kotlin](), [kotlin-coroutines](), [kotlinmultiplatform](), [memory-leaks](), [reactive](), [reactive-streams](), [software](), [suspend](), [swift]() ### AI overview This tutorial explains interoperability between Kotlin coroutines in shared code and RxSwift on iOS. It describes coroutine limitations on Kotlin/Native, including single-threaded release use and experimental multithreading risks, and discusses Swift and Objective-C interop constraints. Example repository code demonstrates single-event and stream operations, success and error cases, and cancellation. ### Source excerpt A recent client engagement involved interop between Kotlin coroutines in shared code, and RxSwift on the iOS side. We did some work to ensure that this could be done in a way that is type-safe and thread-safe. Whether or not you use RxSwift, hopefully this can provide some useful patterns for interop code. Coroutines are a Kotlin language feature that allows asynchronous code to be written in a way that looks like synchronous code, avoiding the nesting that often comes with callback-based APIs. They're available on all Kotlin platforms, but have some limitations on the native side. The release version of native coroutines is limited to single-threaded use-cases, though there are experimental releases available that are multithreaded with some risk of memory leaks. But while they're a fully-supported language feature of Kotlin, they don't translate to Objective-C and Swift. RxSwift is a Swift implementation of the Reactive Streams specification. It's one way to handle asynchronous code on Swift, and has many operators for combining and transforming event streams. Though some of the names are different, much of the API will feel familiar to Kotlin developers who have experience with RxJava. Since Coroutines will almost always be present in shared code, and RxSwift is a common option on the iOS side, hopefully the motivation to communicate between them is clear. So let's start writing some code. Common Repository We'll work with a dummy repository class that looks like this, defined in src/commonMain class ThingRepository { suspend fun getThing(succeed: Boolean): Thing { delay(100) if (succeed) { return Thing(0) } else { error("oh no!") } } fun getThingStream(count: Int, succeed: Boolean): Flow = flow { repeat(count) { delay(100) emit(Thing(it)) } if (!succeed) error("oops!") } } From the outside this looks roughly like a real repository might, but with inputs that let us control the output a bit more directly for demonstration purposes. It gives us the ability ## Using Kotlin Coroutines to Reduce Reactive Code Complexity DevFeed: [Using Kotlin Coroutines to Reduce Reactive Code Complexity]() Original publisher: [Read original article]() Author: piotr-kubowicz Published: 2020-05-12T00:00:00Z Content type: tutorial Language: en Sources: [Backend Development on nexocode]() Topics: [Coroutines](), [Kotlin](), [reactive]() Tags: [coroutines](), [jvm](), [kotlin](), [reactive](), [reactive-programming](), [reactive-streams]() ### AI overview The article compares Kotlin coroutines with reactive programming for backend code. It explains that reactive streams can add complexity and hide empty-result cases, using a Reactor zip example where an empty input can cancel the resulting stream. ### Source excerpt Using Reactive Programming is tempting because it promises to handle higher traffic on the same hardware. Yet this paradigm is also demanding: it increases code complexity and may make bugs harder to notice. Let's explore how you can use Kotlin and coroutines to reduce the cognitive load of maintaining reactive applications. Sample code Imagine you are developing a website that can be used by both logged-in and anonymous users, just like Wikipedia. At the same time, there are proper mechanisms in place to block misbehaving users. Our focus will be the back-end code returning a welcome message displayed on top of the main page: ## From RxJava to Kotlin Flow: Backpressure DevFeed: [From RxJava to Kotlin Flow: Backpressure]() Original publisher: [Read original article]() Published: 2020-02-06T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [backpressure](), [kotlin-flow](), [RxJava](), [Kotlin](), [reactive](), [Reactive Streams]() Tags: [article](), [backpressure](), [coroutines](), [kotlin](), [kotlin-flow](), [reactive-streams](), [rxjava]() ### AI overview This article compares backpressure in RxJava and Kotlin Flow from a black-box perspective. It explains how RxJava Flowable uses Subscriber requests to regulate event production and introduces Kotlin Flow's backpressure solutions for developers considering migration. ### Source excerpt Disclaimer This article unfortunately is not that dive deep (because of complexity of internals of coroutines), but instead trying to compare RxJava and Kotlin Flow mostly as black box with trying to find out differences in behavior and how to migrate from RxJava with its backpressure solution to Kotlin Flow. Hopefully this article will give you direction on where to look at if you decide to migrate to Kotlin Flow. ## Structured Concurrency Anniversary DevFeed: [Structured Concurrency Anniversary]() Original publisher: [Read original article]() Author: Roman Elizarov Published: 2019-09-28T15:02:02Z Content type: article Language: en Sources: [Stories by Roman Elizarov on Medium]() Topics: [Concurrency](), [Kotlin](), [Coroutines](), [kotlin-coroutines](), [Android](), [reactive](), [Documentation]() Tags: [android](), [asynchronous](), [code](), [concurrency](), [coroutine](), [coroutines](), [documentation](), [kotlin](), [kotlin-coroutines](), [programming](), [reactive](), [reactive-streams](), [structured-concurrency]() ### AI overview This retrospective examines the first year of Structured Concurrency in Kotlin Coroutines. It describes its origins in backend asynchronous programming, adoption in Android lifecycle-aware development, and its role in Kotlin Flows and reactive streams. ### Source excerpt Photo by Annie Spratt on Unsplash A little over a year ago I announced big conceptual shift in the design of Kotlin Coroutines called Structured Concurrency. From that moment on, it took our team about a month to make the first stable 1.0.0 release of kotlinx.coroutines library. After a year of further work, kotlinx.coroutines had added stable support for cold flows that integrate nicely with reactive streams. The library had reached version 1.3.2 by now. It is good time to look back and see how it all worked out -- what was great, what could be improved. Structured Concurrency accomplished more than we hoped for. Originally, the design of structured concurrency was based on the woes experienced by backend developers trying to implement all sorts of asynchronous and concurrent logic. It was focused on making sure that you cannot ever lose a running coroutine or an exception. The key building block we added back then is coroutineScope { ... } function, which encapsulates concurrent operations and limits their scope to the scope of the current call. There was not much else to it, so our recommendation to UI developers was to implement CoroutineScope interface in various "closeable" entities of their applications. We envisioned a simple picture with a simple scope hierarchy. It turned out to be more elaborate in practice. Structured Concurrency was rapidly adopted by Android, which has quite complicated life-cycles. Android libraries added extensions like lifecycleScope and viewModelScope, enabling concise and safe integration of coroutines with those concepts. It became apparent that code looks clearer when an object encapsulating the scope is separate from the rest of code. Introductory Android Codelab on Coroutines recommends defining coroutine scope like this: private val scope = CoroutineScope(...) Nowadays, this style increasingly looks more appealing, so it's time to adjust our documentation to reflect it. At the same time, structured concurrency laid a solid fou ## 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 ## 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 ## 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 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 ## RxJava 1 -\> RxJava 2 (Disposing Subscriptions) DevFeed: [RxJava 1 -\> RxJava 2 (Disposing Subscriptions)]() Original publisher: [Read original article]() Author: Kaushik Gopal Published: 2017-06-21T07:00:00Z Content type: tutorial Language: en Sources: [Kaushik Gopal's Site]() Topics: [RxJava](), [reactive](), [Streams]() Tags: [androiddev](), [callback](), [memory-leak](), [reactive-streams](), [rxjava](), [subscriber](), [subscription](), [subscriptions]() ### AI overview This continuation post explains how subscription disposal changed from RxJava 1.x to RxJava 2. In RxJava 2, Reactive Streams interfaces and subscriber callbacks provide access to subscriptions for cancellation and lifecycle management, rather than returning a subscription from subscribe methods. ### Source excerpt 2 part series This is a continuation post in a 2 part series: Understanding the changes Disposing subscriptions Disposing Subscriptions # This was the part that I initially found most tricky to grasp but also most important to know as an AndroidDev (memory leak and all). Jedi master Karnok explains this best in the wiki: In RxJava 1.x, the interface rx.Subscription was responsible for stream and resource lifecycle management, namely unsubscribing a sequence and releasing general resources such as scheduled tasks. The Reactive-Streams specification took this name for specifying an interaction point between a source and a consumer: org.reactivestreams.Subscription allows requesting a positive amount from the upstream and allows cancelling the sequence. From that definition alone, it would appear like nothing's changed but that is definitely not the case. In my first post, I pointed out: Publisher.subscribe(Subscriber) => Subscription The use of => vs = was intentional. If you look at the source code for Publisher's subscribe method again, you'll notice a return type of void viz. it doesn't return a Subscription for you to tack on to a CompositeSubscription (which you can then conveniently dispose of onStop/onDestroy). interface Publisher { // return type void (not Subscription like before) void subscribe(Subscriber s); } Karnok again: Because Reactive-Streams base interface, org.reactivestreams.Publisher defines the subscribe() method as void, Flowable.subscribe(Subscriber) no longer returns any Subscription (or Disposable). The other base reactive types also follow this signature with their respective subscriber types. So if you look at the declarations again // RxJava specific constructs // Observable implements "ObservableSource" interface ObservableSource { void subscribe(Observer observer); } // Single implements SingleSource interface SingleSource { void subscribe(SingleObserver observer); } interface CompletableSource { ## RxJava 1 -\> RxJava 2 (Understanding the changes) DevFeed: [RxJava 1 -\> RxJava 2 (Understanding the changes)]() Original publisher: [Read original article]() Author: Kaushik Gopal Published: 2017-06-21T07:00:00Z Content type: tutorial Language: en Sources: [Kaushik Gopal's Site]() Topics: [RxJava](), [reactive](), [Programming](), [interfaces]() Tags: [classes](), [continuation](), [flowable](), [gradle](), [import](), [interfaces](), [migration](), [programming](), [reactive](), [reactive-programming](), [reactive-streams](), [rxjava](), [series](), [subscriber](), [subscription]() ### AI overview A migration guide for RxJava 1 users explains the major RxJava 2 changes, including its adoption of the Reactive Streams specification, four core interfaces, and package and Gradle dependency changes. ### Source excerpt In case you haven't heard: RxJava2 was released sometime back. RxJava 2 was a massive rewrite with breaking apis (but for good reasons). Most dependent libraries have upgraded by now though, so you're safe to pull that migration trigger with your codebases. Folks starting out directly with Rx2 might enjoy this guide but it's the ones that started with Rx 1 that will probably appreciate it the most. 2 part series This is a continuation post in a 2 part series: Understanding the changes Disposing subscriptions Let's get started. In this first part, I want to dive into making sense of the Rx2 changes from the point of view of an Rx1 user. Why things changed with RxJava2 # tl;dr- Reactive Streams spec ## Reactive Streams is a standard for doing "reactive" programming and RxJava now implements the Reactive Streams specs with version 2.x. RxJava was sort of a trailblazer in reactive programming land but it wasn't the only library around. There were others that also dealt with reactive paradigms. But with all the libraries adhering to the Reactive Streams spec now, interop between the libraries is a tad bit easier. The spec per say is pretty straightforward with just 4 interfaces: Publisher (anything that publishes events, so Observable,Flowable etc. - more on this later) Subscriber (anything that listens to a Publisher) Subscription (Publisher.subscribe(Subscriber) => Subscription when you join a Publisher and a Subscriber, you are given a connection also called a Subscription) Processor (a Publisher + a Subscriber, sound familiar? yep Subjects for us RxJava 1 luddites) If you're slightly more curious about the design goals, I also suggest the following resources: What's different in 2.0 wiki page - this is really the place I kept coming back to and referencing when I needed to understand the details Fragmented Ep #53 with JakeWharton (forgive the shameless promotion) - ultimate lazy person's guide to understand why/what things changed with RxJava2, as explained by an act ## 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 ## 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 ## 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 ## 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