# Flowable Published articles for Flowable. This is one page of public article previews, not the complete archive. Follow Next page to continue. Summaries are not the original full articles. ## Operator fusion in RxJava 2 DevFeed: [Operator fusion in RxJava 2]() Original publisher: [Read original article]() Published: 2019-11-02T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [RxJava](), [Library](), [reactive](), [backpressure]() Tags: [2](), [article](), [backpressure](), [flowable](), [memory](), [overhead](), [performance](), [reactive](), [rxjava](), [technical]() ### AI overview This article explains operator fusion in RxJava 2 as an optimization intended to reduce performance and memory overhead. It first reviews Observable, Flowable, their related observer or subscriber types, and backpressure, then introduces the assembly and subscription stages of an Rx chain. ### Source excerpt Introduction RxJava is a very powerful library, though it has some issues as well. Specifically performance and memory issues, which come from problems library tries to solve and how the solution is designed from technical perspective. In order to minimize overhead in RxJava there are a number of optimizations, which are called "operator fusion". And we'll talk about them in this article. But first let's recap how RxJava reactive types work and what issues they have. ## Flowing in the Deep - Event Streams in Kotlin DevFeed: [Flowing in the Deep - Event Streams in Kotlin]() Original publisher: [Read original article]() Author: Hannes Dorfmann Published: 2019-07-02T00:00:00Z Content type: tutorial Language: en Sources: [Hannes Dorfmann]() Topics: [Kotlin](), [Coroutines](), [RxJava](), [Streams](), [Android]() Tags: [android](), [backpressure](), [coroutine](), [coroutines](), [deep-dive](), [event](), [flow](), [flowable](), [introduction](), [kotlin](), [presentation](), [rxjava](), [scopes](), [streams](), [video]() ### AI overview This presentation introduces Kotlin Flow, compares it with RxJava Observable and Flowable, and explains its relationship to coroutines. It also covers implementation details, custom operators, coroutine channels, backpressure, and coroutine scopes for use in Android apps. ### Source excerpt The Kotlin team introduced a new type called Flow which looks similar to RxJava's Observable or Flowable. Have you ever wondered what's the difference between Flow and RxJava? How does Flow work under the hood? How is it connected to Coroutines? How do you write your own operator? What about Kotlin Coroutine Channels? ## 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 ## 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 ## 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