# Flow API

Published articles for Flow API.

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

## Modernizing Your Android App's Data Storage: SharedPreferences to DataStore

DevFeed: [Modernizing Your Android App's Data Storage: SharedPreferences to DataStore](<https://devfeed.tech/articles/modernizing-your-android-app-s-data-storage-sharedpreferences-to-datastore-25117.md>)

Original publisher: [Read original article](<http://michaelevans.org/blog/2024/10/10/modernizing-your-android-apps-data-storage-sharedpreferences-to-datastore/>)

Author: Michael Evans

Published: 2024-10-10T15:53:57Z

Content type: tutorial

Language: en

Sources: [Gadget Habit](<https://devfeed.tech/sources/gadget-habit.md>)

Topics: [Android](<https://devfeed.tech/topics/android.md>), [migration](<https://devfeed.tech/topics/migration.md>), [schema-evolution](<https://devfeed.tech/topics/schema-evolution.md>), [reactive](<https://devfeed.tech/topics/reactive.md>), [Coroutines](<https://devfeed.tech/topics/coroutines.md>), [Google](<https://devfeed.tech/topics/google.md>)

Tags: [android](<https://devfeed.tech/tags/android.md>), [best-practices](<https://devfeed.tech/tags/best-practices.md>), [coroutines](<https://devfeed.tech/tags/coroutines.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [migration](<https://devfeed.tech/tags/migration.md>), [pitfalls](<https://devfeed.tech/tags/pitfalls.md>), [protocol](<https://devfeed.tech/tags/protocol.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [schema-evolution](<https://devfeed.tech/tags/schema-evolution.md>), [testing](<https://devfeed.tech/tags/testing.md>)

### AI overview

This tutorial explains how to migrate Android app data from SharedPreferences to Proto DataStore. It covers dependencies, schema definition, serialization, repository design, migration testing, schema evolution, error handling, performance, and testing practices.

### Source excerpt

SharedPreferences has long been a staple for storing small pieces of data and user preferences in Android apps. However, it has notable limitations, such as a lack of type safety, no support for safe schema evolution, and potential performance issues on the main thread. Google introduced Proto DataStore as a modern and robust alternative, offering: Strong typing with Protocol Buffers Safe schema evolution Built-in migration support Flow API for reactive programming Coroutines support for main-thread safety In this post, we'll walk through the process of migrating your existing SharedPreferences data to Proto DataStore without data loss, including best practices and common pitfalls to avoid. Step 1: Add Dependencies First, add the necessary dependencies to your app's build.gradle file: 1 2 3 4 5 6 7 8 9 dependencies { def datastore_version = "1.0.0" // Proto DataStore implementation "androidx.datastore:datastore:$datastore_version" // Protocol Buffers implementation "com.google.protobuf:protobuf-javalite:3.18.0" } Step 2: Define Your Proto DataStore Schema Create a new .proto file in app/src/main/proto/my_data.proto to define your data schema: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 syntax = "proto3"; option java_package = "com.example.app"; option java_multiple_files = true; message UserPreferences { // Define your fields with unique numbers string user_name = 1; bool notifications_enabled = 2; string theme = 3; // Optional: Add a version field for future schema evolution int32 schema_version = 999; } Note the schema_version field - this helps manage schema evolution as your app grows. Step 3: Create a Proto DataStore Serializer The serializer handles reading and writing your protocol buffer messages: 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 object UserPreferencesSerializer : Serializer<UserPreferences> { override val defaultValue: UserPreferences = UserPreferences.getDefaultInstance() override suspend fun readFrom(input: InputStream): UserPreferences { try { return UserPrefere

## Java 9 Flow API: Multicasting via a Processor

DevFeed: [Java 9 Flow API: Multicasting via a Processor](<https://devfeed.tech/articles/java-9-flow-api-multicasting-via-a-processor-24818.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-12-14T13:20: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>), [reactive](<https://devfeed.tech/topics/reactive.md>), [Streams](<https://devfeed.tech/topics/streams.md>), [implementation](<https://devfeed.tech/topics/implementation.md>), [API](<https://devfeed.tech/topics/api.md>)

Tags: [backpressure](<https://devfeed.tech/tags/backpressure.md>), [cancellation](<https://devfeed.tech/tags/cancellation.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [multicast](<https://devfeed.tech/tags/multicast.md>), [processor](<https://devfeed.tech/tags/processor.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-streams](<https://devfeed.tech/tags/reactive-streams.md>), [streams](<https://devfeed.tech/tags/streams.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [tck](<https://devfeed.tech/tags/tck.md>)

### 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](<https://devfeed.tech/articles/java-9-flow-api-taking-and-skipping-24812.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-30T20:44:00Z

Content type: tutorial

Language: en

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

Topics: [Java 9](<https://devfeed.tech/topics/java-9.md>), [API](<https://devfeed.tech/topics/api.md>), [reactive](<https://devfeed.tech/topics/reactive.md>)

Tags: [api](<https://devfeed.tech/tags/api.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [implementation](<https://devfeed.tech/tags/implementation.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [publisher](<https://devfeed.tech/tags/publisher.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-streams](<https://devfeed.tech/tags/reactive-streams.md>), [skip](<https://devfeed.tech/tags/skip.md>), [skipwhile](<https://devfeed.tech/tags/skipwhile.md>), [streams](<https://devfeed.tech/tags/streams.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [take](<https://devfeed.tech/tags/take.md>), [takeuntil](<https://devfeed.tech/tags/takeuntil.md>), [takewhile](<https://devfeed.tech/tags/takewhile.md>)

### 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 <T> Flow.Publisher<T> take(Flow.Publisher<T> source, long n) { return new TakePublisher<>(source, n); } The operator's implementation requires little state: static final class TakeSubscriber<T> implements Flow.Subscriber<T> { final Flow.Subscriber<? super T> downstream; Flow.Subscription upstream; long remaining; TakeSubscriber( Flow.Subscriber<? super> 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](<https://devfeed.tech/articles/java-9-flow-api-arbitration-and-concatenation-24807.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-27T12:55:00Z

Content type: tutorial

Language: en

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

Topics: [Java 9](<https://devfeed.tech/topics/java-9.md>), [Java](<https://devfeed.tech/topics/java.md>), [API](<https://devfeed.tech/topics/api.md>)

Tags: [api](<https://devfeed.tech/tags/api.md>), [arbiter](<https://devfeed.tech/tags/arbiter.md>), [arbitration](<https://devfeed.tech/tags/arbitration.md>), [concat](<https://devfeed.tech/tags/concat.md>), [concatenation](<https://devfeed.tech/tags/concatenation.md>), [concurrent](<https://devfeed.tech/tags/concurrent.md>), [event](<https://devfeed.tech/tags/event.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [onerrorresumenext](<https://devfeed.tech/tags/onerrorresumenext.md>), [publisher](<https://devfeed.tech/tags/publisher.md>), [repeat](<https://devfeed.tech/tags/repeat.md>), [request](<https://devfeed.tech/tags/request.md>), [retry](<https://devfeed.tech/tags/retry.md>), [submissionpublisher](<https://devfeed.tech/tags/submissionpublisher.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [subscriptionarbiter](<https://devfeed.tech/tags/subscriptionarbiter.md>)

### 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](<https://devfeed.tech/articles/java-9-flow-api-timing-out-events-24813.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-25T11:46:00Z

Content type: tutorial

Language: en

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

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

Tags: [await](<https://devfeed.tech/tags/await.md>), [concurrency](<https://devfeed.tech/tags/concurrency.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [programming](<https://devfeed.tech/tags/programming.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-programming](<https://devfeed.tech/tags/reactive-programming.md>), [streams](<https://devfeed.tech/tags/streams.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [timeout](<https://devfeed.tech/tags/timeout.md>)

### 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 <T> Flow.Publisher<T> timeout( Flow.Publisher<T> 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](<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

## Java 9 Flow API: ordered merge

DevFeed: [Java 9 Flow API: ordered merge](<https://devfeed.tech/articles/java-9-flow-api-ordered-merge-24810.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-11T14:46: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>)

Tags: [backpressure](<https://devfeed.tech/tags/backpressure.md>), [batching](<https://devfeed.tech/tags/batching.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [merge](<https://devfeed.tech/tags/merge.md>), [ordered-merge](<https://devfeed.tech/tags/ordered-merge.md>), [queue](<https://devfeed.tech/tags/queue.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [sequences](<https://devfeed.tech/tags/sequences.md>), [stream](<https://devfeed.tech/tags/stream.md>)

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

## Java 9 Flow API: mapping asynchronously

DevFeed: [Java 9 Flow API: mapping asynchronously](<https://devfeed.tech/articles/java-9-flow-api-mapping-asynchronously-24809.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-07T14:00: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>), [reactive](<https://devfeed.tech/topics/reactive.md>), [Concurrency](<https://devfeed.tech/topics/concurrency.md>), [implementation](<https://devfeed.tech/topics/implementation.md>), [API](<https://devfeed.tech/topics/api.md>)

Tags: [api](<https://devfeed.tech/tags/api.md>), [asynchronous](<https://devfeed.tech/tags/asynchronous.md>), [concatmap](<https://devfeed.tech/tags/concatmap.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [map](<https://devfeed.tech/tags/map.md>), [mapping](<https://devfeed.tech/tags/mapping.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [varhandle](<https://devfeed.tech/tags/varhandle.md>)

### 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 <T, U> Flow.Publisher<U> mapWhen(Flow.Publisher<T> source, Function<? super T, ? extends Flow.Publisher<U>> mapper) { return mapWhen(source, mapper, (t, u) -> u); } public static <T, U, R> Flow.Publisher<R> mapWhen(Flow.Publisher<T> source, Function<? super T, ? extends Flow.Publisher<U>> mapper, BiFunction<? super T, ? super U, ? extend R> 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

## Java 9 Flow API: mapping and filtering in one stage

DevFeed: [Java 9 Flow API: mapping and filtering in one stage](<https://devfeed.tech/articles/java-9-flow-api-mapping-and-filtering-in-one-stage-24808.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-09-02T14:33:00Z

Content type: tutorial

Language: en

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

Topics: [Java 9](<https://devfeed.tech/topics/java-9.md>), [reactive](<https://devfeed.tech/topics/reactive.md>), [API](<https://devfeed.tech/topics/api.md>), [Library](<https://devfeed.tech/topics/library.md>), [Open Source](<https://devfeed.tech/topics/open-source.md>)

Tags: [backpressure](<https://devfeed.tech/tags/backpressure.md>), [exception](<https://devfeed.tech/tags/exception.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [publisher](<https://devfeed.tech/tags/publisher.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-streams](<https://devfeed.tech/tags/reactive-streams.md>), [rxjava](<https://devfeed.tech/tags/rxjava.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>)

### 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<String> 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](<https://devfeed.tech/articles/java-9-flow-api-asynchronous-integer-range-source-24805.md>)

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

Author: David Karnok (noreply@blogger.com)

Published: 2017-03-05T13:08: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>), [reactive](<https://devfeed.tech/topics/reactive.md>), [interfaces](<https://devfeed.tech/topics/interfaces.md>), [IntelliJ IDEA](<https://devfeed.tech/topics/intellij-idea.md>)

Tags: [asynchronous](<https://devfeed.tech/tags/asynchronous.md>), [backpressure](<https://devfeed.tech/tags/backpressure.md>), [coroutine](<https://devfeed.tech/tags/coroutine.md>), [flow](<https://devfeed.tech/tags/flow.md>), [flow-api](<https://devfeed.tech/tags/flow-api.md>), [java](<https://devfeed.tech/tags/java.md>), [java-9](<https://devfeed.tech/tags/java-9.md>), [jdk](<https://devfeed.tech/tags/jdk.md>), [operator](<https://devfeed.tech/tags/operator.md>), [publisher](<https://devfeed.tech/tags/publisher.md>), [range](<https://devfeed.tech/tags/range.md>), [reactive](<https://devfeed.tech/tags/reactive.md>), [reactive-streams](<https://devfeed.tech/tags/reactive-streams.md>), [request-management](<https://devfeed.tech/tags/request-management.md>), [source](<https://devfeed.tech/tags/source.md>), [submissionpublisher](<https://devfeed.tech/tags/submissionpublisher.md>), [subscriber](<https://devfeed.tech/tags/subscriber.md>), [subscription](<https://devfeed.tech/tags/subscription.md>), [trampoline](<https://devfeed.tech/tags/trampoline.md>), [varhandle](<https://devfeed.tech/tags/varhandle.md>)

### 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<Integer> { 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<? super Integer> subscriber) { // TODO implement } } For brevit