# backpressure Published articles for backpressure. This is one page of public article previews, not the complete archive. Follow Next page to continue. Summaries are not the original full articles. ## Worker Backpressure (Part 1) DevFeed: [Worker Backpressure (Part 1)]() Original publisher: [Read original article]() Author: Mikalai Barysau Published: 2026-09-17T00:00:00Z Content type: article Language: en Sources: [Canva Engineering]() Topics: [backpressure](), [Back end](), [reliability]() Tags: [backend](), [backpressure](), [dependencies](), [outage](), [queue](), [reliability](), [worker]() ### AI overview Canva describes Worker Backpressure, a reliability mechanism for queue workers that slows processing when dependencies fail and speeds it back up after recovery. During a cloud-provider outage, the mechanism helped prevent failed messages from accumulating on the Dead Letter Queue and avoided an on-call page. ### Source excerpt How we taught Canva's queue workers to slow down when dependencies fail, then speed back up on their own. ## Broadcasting to 28K SSE Clients (and learning about Go channels) DevFeed: [Broadcasting to 28K SSE Clients (and learning about Go channels)]() Original publisher: [Read original article]() Author: Pranshu Raj Published: 2026-07-23T00:00:00Z Content type: tutorial Language: en Sources: [Pranshu Raj - blog on backend systems, performance and sidequests]() Topics: [Go Language](), [Server-sent events (SSE)](), [Concurrency](), [backpressure](), [real-time](), [observability](), [Redis](), [Grafana](), [Prometheus]() Tags: [architecture](), [backpressure](), [concurrency](), [go](), [grafana](), [observability](), [prometheus](), [real-time](), [redis](), [sse]() ### AI overview The article describes building a real-time leaderboard in Go that reached 28,232 concurrent SSE connections before exposing a broken broadcast design. It explains the fix using centralized polling, deduplication, fan-out, backpressure handling, and observability with Prometheus and Grafana. ### Source excerpt A post on progressively getting my server to 28k SSE connections, finding a bug in my implementation of fan out using Go channels and the resulting mental model update. ## Backpressure in Distributed Systems DevFeed: [Backpressure in Distributed Systems]() Original publisher: [Read original article]() Author: Pranshu Raj Published: 2026-07-23T00:00:00Z Content type: tutorial Language: en Sources: [Pranshu Raj - blog on backend systems, performance and sidequests]() Topics: [backpressure](), [distributed-systems](), [systems](), [Messaging](), [real-time]() Tags: [backpressure](), [distributed](), [distributed-systems](), [go](), [latency](), [memory](), [messages](), [messaging](), [real-time](), [systems]() ### AI overview This tutorial explains backpressure in distributed systems, including how it occurs when message producers outpace consumers and the resulting risks such as memory exhaustion, dropped messages, lower throughput, network waste, increased latency, and blocked producers. It introduces producer, messaging-system, and consumer roles and relates the topic to a real-time leaderboard project. ### Source excerpt Understanding what it is, how to deal with it, where it's used and how I handled it in the real time leaderboard. ## How We Cut Kafka Consumer Deployment Costs by 83% DevFeed: [How We Cut Kafka Consumer Deployment Costs by 83%]() Original publisher: [Read original article]() Author: ZhongLi Shen He creates fantastic software as a backend engineer GitHub profile Linkedin profile Published: 2026-06-12T00:00:00Z Content type: tutorial Language: en Sources: [Trivago]() Topics: [Kafka](), [Deployer](), [Deployment](), [reactive](), [Library]() Tags: [backend](), [backpressure](), [debugging](), [deployment](), [engineering](), [infrastructure](), [java](), [kafka](), [library](), [performance](), [reactive]() ### AI overview This post explains how trivago investigated PSE-kafka's low CPU usage, slow Kafka polling, and consumer lag. The investigation found that replacing the in-house KafkaReceiverFlux with spring-kafka removed the lag, while the team continued looking for a suitable fix that preserved pull-based consumption and backpressure. The changes ultimately cut infrastructure costs by 83% and ended a run of 19 P1 incidents. ### Source excerpt This post walks through a layered performance investigation that cut PSE-kafka's infrastructure costs by 83% and ended a run of 19 P1 incidents.PSE-kafka (price-search-engine-kafka) i... ## Backpressure improvements help prevent data loss during ingestion traffic spikes DevFeed: [Backpressure improvements help prevent data loss during ingestion traffic spikes]() Original publisher: [Read original article]() Author: Javier Goizueta Published: 2025-11-18T12:00:00Z Content type: article Language: en Sources: [Tinybird]() Topics: [data]() Tags: [backpressure](), [data](), [engineering-excellence]() ### AI overview The article explains that backpressure improvements help prevent data loss during traffic spikes and keep ingestion reliable when volume exceeds expectations. ### Source excerpt Backpressure improvements prevent data loss during traffic spikes. Ingestion stays reliable even when volume exceeds expectations. ## Fearless Concurrency Ep.7: Lock-Free Structures and Channels for Scalable Rust Code DevFeed: [Fearless Concurrency Ep.7: Lock-Free Structures and Channels for Scalable Rust Code]() Original publisher: [Read original article]() Published: 2024-12-05T00:00:00Z Content type: tutorial Language: en Sources: [William Kennedy]() Topics: [Concurrency](), [Rust](), [Data structures](), [Code](), [systems]() Tags: [advanced-rust-concurrency](), [alternatives](), [backpressure](), [concurrency](), [concurrency-in-rust](), [crossbeam-crate-rust](), [efficient-multithreading-rust](), [fearless-concurrency-rust](), [lock-free](), [lock-free-data-structures-rust](), [managing-shared-resources-rust](), [mpsc-channels-rust](), [one-shot-channels-rust](), [optimizing-concurrency-in-rust](), [performance](), [robust-multithreaded-rust-apps](), [rust](), [rust-atomiccell](), [rust-bounded-channels](), [rust-concurrency](), [rust-dashmap-dashset](), [rust-high-concurrency-techniques](), [rust-thread-multiplexing](), [rust-thread-safe-work-queues](), [rust-timeout-options](), [scalable-rust-applications](), [thread](), [thread-communication-rust]() ### AI overview The seventh and final episode of a Rust concurrency series covers lock-free data structures, channels for communication between threads, and techniques for managing shared resources. It discusses tools including DashMap, DashSet, Crossbeam, MPSC and one-shot channels, multiplexing, bounded channels, backpressure, memory usage, and timeouts. ### Source excerpt Introduction: Welcome to Episode 7 of the Fearless Concurrency in Rust series! In this final episode, we explore advanced concurrency techniques that enable efficient, scalable, and robust multithreaded applications in Rust. The focus is on leveraging tools like lock-free data structures, channels for thread communication, and strategies for safely managing shared resources in complex systems. These approaches ensure developers can push the limits of Rust's concurrency model while maintaining safety and performance. ## Efficient data handling with the Streams API DevFeed: [Efficient data handling with the Streams API]() Original publisher: [Read original article]() Author: vultr Published: 2024-09-06T00:00:00Z Content type: article Language: en Sources: [MDN Blog]() Topics: [Streaming](), [JavaScript](), [web applications](), [API](), [Web Development](), [Command-line interface](), [Firewall](), [ssh]() Tags: [api](), [backpressure](), [data](), [efficiency](), [how-to](), [javascript](), [performance](), [stream](), [streaming](), [streams](), [web-applications]() ### AI overview This article explains how the Streams API lets JavaScript process network data incrementally as it arrives, instead of waiting for an entire resource to download. It covers chunk-based processing, backpressure, piping, and stream abstractions, then demonstrates a web application that transforms a data stream. ### Source excerpt This post demonstrates how to use the Streams API in a web application to fetch and transform text on the fly. By processing the data as it arrives, this approach enhances performance, responsiveness, and resource efficiency. ## A Pattern for Composing Flow Operations DevFeed: [A Pattern for Composing Flow Operations]() Original publisher: [Read original article]() Published: 2024-04-15T00:00:00Z Content type: tutorial Language: en Sources: [Kt. Academy]() Topics: [Kotlin](), [kotlin-coroutines](), [Streams](), [async](), [backpressure](), [flux](), [RxJava]() Tags: [async](), [backpressure](), [flux](), [jpa](), [kotlin](), [kotlin-coroutines](), [rxjava](), [streams](), [workshop-learning-programming]() ### AI overview This tutorial explains a pattern for composing Kotlin Flow operations by collecting one cold flow within another flow definition. It discusses sequential processing, asynchronous operations, and the limitations of using internal flow types when composing streams. ### Source excerpt Extending and operating on top of your existing flows ## 3 (More) Tips for Optimizing Apache Flink Applications DevFeed: [3 (More) Tips for Optimizing Apache Flink Applications]() Original publisher: [Read original article]() Author: Kevin Lam Published: 2022-12-13T14:00:00Z Content type: tutorial Language: en Sources: [Shopify Engineering](), [Shopify Engineering - Shopify Engineering]() Topics: [apache-flink](), [data-processing](), [Streaming](), [data]() Tags: [apache-flink](), [backpressure](), [best-practices](), [parallelism](), [performance](), [shopify](), [stateful]() ### AI overview Shopify presents additional best practices for optimizing large, stateful Apache Flink applications. The supplied text covers selecting appropriate parallelism to scale execution and improve resource utilization, and avoiding bottlenecks in data sinks that can reduce throughput and cause backpressure. ### Source excerpt Earlier this year, we shared our tips for optimizing large stateful Apache Flink applications. Below we'll walk you through 3 more best practices. ## Integrating Confluent Schema Registry with Apache Spark applications DevFeed: [Integrating Confluent Schema Registry with Apache Spark applications]() Original publisher: [Read original article]() Author: Dominik Liebler Published: 2022-01-24T08:04:19Z Content type: tutorial Language: en Sources: [YAZIO Engineering - Medium]() Topics: [Kafka](), [Apache Spark](), [Streaming](), [data lake](), [parquet](), [Kotlin](), [Kubernetes](), [ceph](), [JSON Schema]() Tags: [apache-spark](), [backpressure](), [ceph](), [confluent](), [data-engineering](), [data-lake](), [data-pipeline](), [json](), [kafka](), [kotlin](), [kubernetes](), [parquet](), [payload](), [pipelines](), [schema](), [schemaregistry](), [serialization](), [spark](), [streaming]() ### AI overview This engineering article explains YAZIO's data pipeline from mobile and web applications through Kafka and Spark Structured Streaming into a Ceph-based data lake. It discusses why schemas matter and describes replacing JSON with Apache Avro and Confluent Schema Registry to reduce message size while keeping schema information externally stored and cached. ### Source excerpt At YAZIO, we believe in making decisions backed by data to help people live healthier lives through better nutrition. For each new and existing feature we want to evaluate how well it performs and how our users interact with it. In order to do so, we need a lot of data and we need to handle backpressure in our systems. To cope with that we use a Kafka cluster managed by Strimzi operators running in Kubernetes. The data itself is being ingested from our mobile and web apps via HTTP or TCP endpoints serialized into JSON and stored in Kafka by a small application written in Kotlin/JVM. Overview of our data pipeline architecture At the other end of the pipeline, different Spark Structured Streaming applications (also written in Kotlin) dump this information into our data lake residing in a Ceph bucket. They read data from Kafka, deserialize it, transform some of the fields and write Parquet files into the data lake using a new schema. Why schemas? Schemas play an important role in data pipelines because they give meaning and context to data. In a world without schemas we would still do random interpretations about the context and meaning of data every now and then when using it. As you might have guessed already this would lead to a lot of bugs and misunderstandings. Photo by EJ Strat https://unsplash.com/photos/VjWi56AWQ9k Similar to a legal contract that binds you to certain limits, a schema binds the data to certain limits and meaning which narrow down the need of interpretation. Choice of serialization formats At the time of writing, Confluent Schema Registry supports these three serialization formats: Apache Avro Protocol Buffers (protobuf) JSON Schema From those choices, only two really provide more than just validation of the data that is ingested and transmitted through our data pipelines. Avro and Protobuf also allow us to shrink the sizes of our topics because only the payload is contained in a message, while the repeating schema will not be stored. In the cas ## Java Reactive Programming - Effective Usage in a Real World Application DevFeed: [Java Reactive Programming - Effective Usage in a Real World Application]() Original publisher: [Read original article]() Author: Stefan Nothaas Java Software; Backend Engineer Published: 2021-03-16T00:00:00Z Content type: article Language: en Sources: [Trivago]() Topics: [Java](), [reactive](), [Programming](), [Back end](), [Microservice](), [Kafka](), [API]() Tags: [api](), [backend](), [backpressure](), [blocking](), [java](), [kafka](), [reactive](), [reactive-programming](), [services]() ### AI overview This article explains how trivago's search backend team applied reactive programming in Java to a production microservice. It covers handling blocking I/O, backpressure, varying loads, message and error propagation, and implementation challenges. ### Source excerpt This article presents how trivago's search backend team used reactive programming in Java effectively when designing and implementing one of our many Java backend services. Compared to tradi... ## From RxJava to Kotlin Flow: Throttling DevFeed: [From RxJava to Kotlin Flow: Throttling]() Original publisher: [Read original article]() Published: 2020-03-02T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [reactive](), [kotlin-flow](), [RxJava](), [Programming]() Tags: [backpressure](), [kotlin](), [kotlin-flow](), [reactive](), [reactive-programming](), [rxjava]() ### AI overview A comparison of throttling operators in RxJava and Kotlin Flow, explaining how debounce, throttleLast or sample, and throttleFirst limit events in reactive streams. The article uses UI interactions such as search input and button clicks as examples. ### Source excerpt Source This post is part of series of comparing RxJava to Kotlin Flow. Previous articles were about Threading, Backpressure, Error Handling, Stream Types It is advised to take a look at the overview of Throttling operators in RxJava before reading this article: Throttling in RxJava 2 Introduction Throttling is also one of the essential features of reactive programming. Streams might emit more values than we're expecting to receive, so ability to limit number of events in a stream is very important. Most common place where it applies is handling UI events. Examples of usage might be handling button clicks (to not allow double clicks), usage with search view when we'd like to wait for user to stop typing before querying some data and so on. In this article we'll try to compare throttling operators in RxJava and Kotlin Flow. ## From RxJava to Kotlin Flow: Stream Types DevFeed: [From RxJava to Kotlin Flow: Stream Types]() Original publisher: [Read original article]() Published: 2020-02-26T00:00:00Z Content type: tutorial Language: en Sources: [Vasya Drobushkov]() Topics: [kotlin-flow](), [RxJava](), [Programming](), [reactive](), [backpressure]() Tags: [backpressure](), [flow](), [kotlin](), [kotlin-flow](), [reactive](), [reactive-programming](), [rxjava]() ### AI overview This article compares stream types in RxJava and Kotlin Flow as part of a migration series. It explains RxJava's Observable, Flowable, Single, Maybe, and Completable, then maps them to Kotlin Flow or suspending functions and highlights differences in backpressure and stream composition. ### Source excerpt Source This post is part of series of comparing RxJava to Kotlin Flow. Previous articles were about Threading, Backpressure, Error Handling Introduction In reactive programming we use streams. Therefore in both RxJava and Kotlin Flow we'll have to use some streams. Though we'll encounter some differences if we decide to migrate. In this article we'll try to get into the stream types which have RxJava and Kotlin Flow and some important differences. ## 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. ## 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? ## 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 ## 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 ## Java 9 Flow API: ordered merge DevFeed: [Java 9 Flow API: ordered merge]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-11T14:46:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [API]() Tags: [backpressure](), [batching](), [flow](), [flow-api](), [java](), [java-9](), [merge](), [ordered-merge](), [queue](), [reactive](), [rxjava](), [sequences](), [stream]() ### AI overview This article explains why zip() and flatMap() are unsuitable for merging multiple ordered event sequences while preserving order. It presents an orderedMerge() operator that selects the smallest or largest available item, requires a fixed number of source sequences, and discusses how unordered sources produce priority-queue-like output. It also introduces prefetching, queues, coordination, batching, and stable-prefetch backpressure for an inner consumer implementation. ### Source excerpt Introduction Sometimes, one has several ordered sequences of events and would like to merge them into one single flow. Since one element from a sequence should come before another element in another sequence, we need a way to keep comparing elements with each other from different sequences. Unfortunately, zip() doesn't work because it takes a row of available items and item #2 from sequence #2 may come before item #1 from stream #3. Plus, if one stream is shorter than the others, the end sequence stops. Similarly, flatMap() doesn't work because it takes the next item from any inner source sequence the moment it is available without any ordering considerations at that point. At least it emits all items from all sources (provided there are no errors of course). Therefore, we need something between the two operators: one that collects up a row of items from the sources, decides which is the smallest/largest of them based on some comparison logic and only emits that. It then awaits a fresh item from that specific source (or completion) and repeats the picking of the smallest/largest item as long as there are requests for it. Such operator, let's call it orderedMerge(), has an implication about the number of its inner source sequences: it has to be fixed. The reason for it is that it has to pick the smallest/largest of the available items in order for the output to be in order. If there is still a source missing, it can't know for sure the others are smaller/larger that any of the upcoming item from that missing source will produce. The second implication is, what happens if the sources themselves are not ordered? The logic presented in this post still works, but the end output won't be totally ordered. It will act like some priority queue instead: picking important items first before turning to less important ones. The inner consumer Operators handling multiple sources often need a way to prefetch item from these sources and give out them on demand to some joining logic ## Java 9 Flow API: mapping and filtering in one stage DevFeed: [Java 9 Flow API: mapping and filtering in one stage]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-09-02T14:33:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java 9](), [reactive](), [API](), [Library](), [Open Source]() Tags: [backpressure](), [exception](), [flow](), [flow-api](), [java-9](), [publisher](), [reactive](), [reactive-streams](), [rxjava](), [subscriber]() ### AI overview This article examines how to combine mapping and filtering into a single custom operator for Java 9 Flow API publishers. It discusses API design choices, handling mapped, dropped, failed, and completed items, and preserving Reactive Streams backpressure and protocol requirements. ### Source excerpt Introduction In most reactive libraries, mapping and filtering can be done on a flow via separate operators map() and filter() respectively. One rare occasions, the functions to these operators would need to communicate with each other without sharing information in a flow-external manner and without using defer(). Such combined and standard mapFilter() operator doesn't exist and one has to write one of its own. Given that the Java 9 Flow API is brand new, one has to definitely write a custom operator for it as Java 9 itself doesn't provide any rich set of predefined operations on Flow.Publishers unlike its dual, the Stream API. Shameless advertising By the way, if you are looking for a Java 9 Flow-based, native and modern reactive library with rich set of operators, similar to RxJava 2 (even including some operators from its extension project), I happen to have one for you: Reactive4JavaFlow. It is free and open-source with the promising outlook that one day, it may form the basis for the next major RxJava version... MapFilter API design When the Reactive4Java library was first concieved in 2011, the first significant stumbling block was not the lack of lambdas in Java 6/7 but the lack of extension methods. C# had it and made Rx.NET conveniently extendable (assuming you managed to understand how to write operators for it as it wasn't open source at the time). Java still doesn't have any sign of ever getting extension methods, therefore, we either need a rich abstract base class, such as Flowable or Flux, or an utility class whose methods almost look like extension method definitions with the exception that the developer has to stack them on top of one another: import java.util.concurrent.*; import static FlowUtils.*; Flow.Publisher f = timeout( mapFilter( new FlowRange(1, 10, Runnable::run), (v, e) -> { if (v % 2 == 0) { e.next(v.toString()) } if (v == 7) { e.complete(); } } ), 5, TimeUnit.MILLISECONDS ); When thinking about a combined map and filter operat ## Java 9 Flow API: asynchronous integer range source DevFeed: [Java 9 Flow API: asynchronous integer range source]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2017-03-05T13:08:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Java](), [Java 9](), [reactive](), [interfaces](), [IntelliJ IDEA]() Tags: [asynchronous](), [backpressure](), [coroutine](), [flow](), [flow-api](), [java](), [java-9](), [jdk](), [operator](), [publisher](), [range](), [reactive](), [reactive-streams](), [request-management](), [source](), [submissionpublisher](), [subscriber](), [subscription](), [trampoline](), [varhandle]() ### AI overview A tutorial explores Java 9's Flow API and Reactive Streams interfaces by building an asynchronous integer range Publisher. It discusses composing Publishers, implementing Flow.Subscription, handling subscriber demand, and using IntelliJ 2017.1 EAP while the APIs were non-final. ### Source excerpt Introduction Java 9 is becoming more reactive by introducing the Reactive-Streams interfaces under the parent class java.util.concurrent.Flow, enabling a new standard interoperation between future libraries built on top. There is almost no documentation beyond a underwhelming Oracle documentation and the SubmissionPublisher class' JavaDoc about how to write Publishers, Subscriptions and Subscribers under the Flow API. Plus the Oracle document practically concludes with see RxJava. Indeed, replacing the imports of org.reactivestreams.* with java.util.concurrent.Flow.* in RxJava 2's sources get's one a fully fledged reactive library but there seems to be one crucial expectation with components built on the Flow API: they have to be asynchronous at every stage. I could argue that the underlying concepts work totally fine in synchronous mode, but who am I to question the established definitions? Oh well, if the constraint is to be asynchronous, then let's do it in an asynchronous way. To see what it takes, we could start with a relatively simple source: an asynchronous integer range. Since both Java 9 and the IDE support is in non-final state, I recommend IntelliJ 2017.1 EAP for this "exercise". Asynchronous integer range source Unfortunately, Java 9 won't introduce any standard fluent API entry point with all the well loved map(), filter(), flatMap() etc. operators but one has to build individual Publishers and compose them stage-by-stage. This involves creating a parent Publisher class with the following typical pattern to host the input parameters of the flow to be observed: import java.util.concurrent.*; public final class FlowRange implements Flow.Publisher { final int start; final int end; final Executor executor; public FlowRange(int start, int count, Executor executor) { this.start = start; this.end = start + count; this.executor = executor; } @Override public void subscribe(Flow.Subscriber subscriber) { // TODO implement } } For brevit ## 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 ## SubscribeOn and ObserveOn DevFeed: [SubscribeOn and ObserveOn]() Original publisher: [Read original article]() Author: David Karnok (noreply@blogger.com) Published: 2016-03-31T12:55:00Z Content type: tutorial Language: en Sources: [Akarnokd - Advanced RxJava]() Topics: [Programming](), [reactive](), [User Experience](), [Android](), [GUI]() Tags: [backpressure](), [blocking](), [cancellation](), [executorservice](), [main-thread](), [observeon](), [reactive](), [rxjava](), [subscribeon](), [thread]() ### AI overview A tutorial explaining why RxJava's subscribeOn and observeOn operators are often confused. It distinguishes their effects by tracing subscription and method-call flow, and discusses moving subscription side effects such as network, database, or blocking work away from the current thread. ### Source excerpt Introduction One of the most confused operator pair of the reactive ecosystem is the subscribeOn and observeOn operators. The source of confusion may be rooted in a few causes: they sound alike, they sometimes show similar behavior when looked at from downstream and they are duals in some sense. It appears the name-confusion isn't local to RxJava. Project Reactor faces a similar issue with their publishOn and dispatchOn operators. Apparently, it doesn't matter what they are called and people will confuse them anyhow. When I started learning about Rx.NET back in 2010, I never experienced this confusion; subscribeOn affects subscribe() and observeOn affects onXXX(). (Remark: I've searched Channel 9 for the early videos but couldn't really find the talk where they build up these operators just like I'm about to do. The closest thing was this.) My "thesis" is that the confusion may be resolved by walking through how one can implement these operators and thus showing the internal method-call flow. SubscribeOn The purpose of subscribeOn() is to make sure side-effects from calling subscribe() happens on some other thread. However, almost no standard RxJava source does side-effects on its own; you can have side-effects with custom Observables, wrapped subscription-actions via create() or as of lately, the with the SyncOnSubscribe and fromCallable() APIs. Why would one move the side-effects? The main use cases are doing network calls or database access on the current thread or anything that involves blocking wait. Holding off a Tomcat worker thread hasn't been much of a programming problem (that doesn't mean we can't improve the stack with reactive) but holding off the Event Dispatch Thread in a Swing application or the Main thread in an Android application has adverse effect on the user experience. (Sidenote: it's a funny thing that blocking the EDT is basically a convenience backpressure strategy in the GUI world to prevent the user from changing the application state whil [Next page]()