# observeOn

Published articles for observeOn.

This is one page of public article previews, not the complete archive. Follow Next page to continue. Summaries are not the original full articles.

## Unit Tests and Concurrency

DevFeed: [Unit Tests and Concurrency](<https://devfeed.tech/articles/unit-tests-and-concurrency-25874.md>)

Original publisher: [Read original article](<http://lordraydenmk.github.io//2021/unit-tests-and-concurrency/>)

Author: Stojan Anastasov

Published: 2021-01-06T00:00:00Z

Content type: tutorial

Language: en

Sources: [Stojan Anastasov's blog](<https://devfeed.tech/sources/stojan-anastasov-s-blog.md>)

Topics: [RxJava](<https://devfeed.tech/topics/rxjava.md>), [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [Android](<https://devfeed.tech/topics/android.md>), [Testing](<https://devfeed.tech/topics/testing.md>)

Tags: [android](<https://devfeed.tech/tags/android.md>), [api](<https://devfeed.tech/tags/api.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [junit](<https://devfeed.tech/tags/junit.md>), [observeon](<https://devfeed.tech/tags/observeon.md>), [recyclerview](<https://devfeed.tech/tags/recyclerview.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [scheduler](<https://devfeed.tech/tags/scheduler.md>), [testing](<https://devfeed.tech/tags/testing.md>), [tests](<https://devfeed.tech/tags/tests.md>), [trampoline](<https://devfeed.tech/tags/trampoline.md>)

### AI overview

This tutorial explains how replacing RxJava schedulers with a single test scheduler can hide concurrency problems. It presents an Android example involving concurrent API calls and discusses refactoring the RxJava chain to update the UI as results become available.

### Source excerpt

Once Retrofit added RxJava support, RxJava became my go-to concurrency framework for writing Android apps. One of the great things about RxJava is the excellent testing support. It includes TestObserver, TestScheduler, RxJavaPlugins so you can switch your schedulers in tests. A common approach in testing RxJava code is using a JUnit rule that replaces the Scheduler pools with Schedulers.trampoline() before tests are run and resets them to the original thread pools after the tests. This makes the whole Observable chain runs on a single thread, the same thread the test runs on, which means we can write assertions without worrying about concurrency. However the production code usually is not single threaded. IO operations are done on the IO thread pool, views are updated on the main thread and everything else happens on the computation pool. By using different schedulers in the tests and using a different strategy (single threaded) we make those unit tests useless in catching concurrency issues. A real world scenario I was working on a side project. The screen consists of a RecyclerView displaying a list of elements. To get the elements I need to perform two different API calls. The first API call returns a list with N elements, then for each item in the list I need to perform the second call. After combining the data I send it to the UI for displaying. Using RxJava this looks like: // Emits Loading then Content or Problem private fun requestData(): Observable<ViewState> = service.firstApiCall() .observeOn(Schedulers.computation()) .map { it.message } .flatMap(this::secondApiCall) .map<ViewState> { ViewState.Content(it) } .startWith(Single.just(ViewState.Loading)) .onErrorReturn { ViewState.Problem } .toObservable() // Concurrently executes secondApiCall for each element in list. // Transforms the result to ViewEntity, combines everything in a list private fun secondApiCall(list: List<String>): Single<List<ViewEntity>> = Observable.fromIterable(list) .concatMapEager {

## From RxJava 2 to Kotlin Flow: Threading

DevFeed: [From RxJava 2 to Kotlin Flow: Threading](<https://devfeed.tech/articles/from-rxjava-2-to-kotlin-flow-threading-38624.md>)

Original publisher: [Read original article](<https://krossovochkin.com/posts/2020_01_25_from_rxjava_2_to_kotlin_flow_threading/>)

Published: 2020-01-25T00:00:00Z

Content type: tutorial

Language: en

Sources: [Vasya Drobushkov](<https://devfeed.tech/sources/vasya-drobushkov.md>)

Topics: [RxJava](<https://devfeed.tech/topics/rxjava.md>), [kotlin-flow](<https://devfeed.tech/topics/kotlin-flow.md>), [Kotlin](<https://devfeed.tech/topics/kotlin.md>), [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [reactive](<https://devfeed.tech/topics/reactive.md>), [Parallelism](<https://devfeed.tech/topics/parallelism.md>), [Android](<https://devfeed.tech/topics/android.md>), [Programming](<https://devfeed.tech/topics/programming.md>)

Tags: [android](<https://devfeed.tech/tags/android.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [kotlin](<https://devfeed.tech/tags/kotlin.md>), [kotlin-flow](<https://devfeed.tech/tags/kotlin-flow.md>), [observeon](<https://devfeed.tech/tags/observeon.md>), [parallelism](<https://devfeed.tech/tags/parallelism.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-programming](<https://devfeed.tech/tags/reactive-programming.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [streams](<https://devfeed.tech/tags/streams.md>), [subscribeon](<https://devfeed.tech/tags/subscribeon.md>)

### AI overview

This article compares threading in RxJava 2 and Kotlin Flow. It explains cold streams and the roles of subscribeOn and observeOn, including how schedulers determine where a reactive chain starts and continues.

### Source excerpt

Source Introduction For a long time RxJava was undisputed leader for reactive solutions on Android, though with Kotlin expansion and introducing cold streams (Flow) seems situation might be rapidly changing in the coming years. Though reactive programming is not related to threading in the first place, concurrency and parallelism are very important anyway. In this article we'll try to make short recap on threading in RxJava 2 (with some basic caveats on its usage) and then take a look at how threading works in Kotlin Flow, so if anyone would like to migrate their code without affecting functionality it would be nice and smooth.

## Java 9 Flow API: switching threads

DevFeed: [Java 9 Flow API: switching threads](<https://devfeed.tech/articles/java-9-flow-api-switching-threads-24811.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2017/09/java-9-flow-api-switching-threads.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-21T10:49:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [Java](<https://devfeed.tech/topics/java.md>), [Java 9](<https://devfeed.tech/topics/java-9.md>), [API](<https://devfeed.tech/topics/api.md>), [reactive](<https://devfeed.tech/topics/reactive.md>), [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [Concurrent Programming](<https://devfeed.tech/topics/concurrent-programming.md>)

Tags: [asynchronous](<https://devfeed.tech/tags/asynchronous.md>), [executor](<https://devfeed.tech/tags/executor.md>), [executorservice](<https://devfeed.tech/tags/executorservice.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [idea](<https://devfeed.tech/tags/idea.md>), [intellij](<https://devfeed.tech/tags/intellij.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jvm](<https://devfeed.tech/tags/jvm.md>), [main-thread](<https://devfeed.tech/tags/main-thread.md>), [observeon](<https://devfeed.tech/tags/observeon.md>), [publisher](<https://devfeed.tech/tags/publisher.md>), [thread](<https://devfeed.tech/tags/thread.md>), [threading](<https://devfeed.tech/tags/threading.md>), [threads](<https://devfeed.tech/tags/threads.md>)

### 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 <T> Flow.Publisher<T> observeOn( Flow.Publisher<T> 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<T> implements Flow.Subscriber<T>, Flow.Subscription, Runnable { final Flow.Subscriber<? super T> downstream; final Executor exec; final int prefetch; final Queue<T> queue; F

## SubscribeOn and ObserveOn

DevFeed: [SubscribeOn and ObserveOn](<https://devfeed.tech/articles/subscribeon-and-observeon-24799.md>)

Original publisher: [Read original article](<https://akarnokd.blogspot.com/2016/03/subscribeon-and-observeon.html>)

Author: David Karnok (noreply@blogger.com)

Published: 2016-03-31T12:55:00Z

Content type: tutorial

Language: en

Sources: [Akarnokd - Advanced RxJava](<https://devfeed.tech/sources/akarnokd-advanced-rxjava.md>)

Topics: [Programming](<https://devfeed.tech/topics/programming.md>), [reactive](<https://devfeed.tech/topics/reactive.md>), [User Experience](<https://devfeed.tech/topics/user-experience.md>), [Android](<https://devfeed.tech/topics/android.md>), [GUI](<https://devfeed.tech/topics/gui.md>)

Tags: [backpressure](<https://devfeed.tech/tags/backpressure.md>), [blocking](<https://devfeed.tech/tags/blocking.md>), [cancellation](<https://devfeed.tech/tags/cancellation.md>), [executorservice](<https://devfeed.tech/tags/executorservice.md>), [main-thread](<https://devfeed.tech/tags/main-thread.md>), [observeon](<https://devfeed.tech/tags/observeon.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [subscribeon](<https://devfeed.tech/tags/subscribeon.md>), [thread](<https://devfeed.tech/tags/thread.md>)

### 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