Operators on Flows

Flow is stream of data. Sometime we want to alter the data stream by mapping, merging, filtering, sorting, reducing, splitting, chunking, windowing, sampling, and flat-mapping.

For that, there are a series of extension functions on Flow called “operators”.

We will try to dive into few of popular Flow operators, and understand their use case.

debounce(): Filtering Fast Flows

val slowFlow = originalFlow.debounce(250)

In debounce(), originalFlow is the input stream and slowFlow is the output stream. Sometimes, you have a Flow that emits too quickly, and you want only those values that have not changed in a while. here debounce acts as a saviour.

Consider the scenario of user input in a text field. Normally, you would want to use that input to filter data from a database or server as the user types. However, since I/O operations can be slow, updating the filter after every keypress may not be ideal. Instead, you would prefer to wait until the user has finished typing, even though there may not be any explicit indication of that. So, the approach is to wait for a significant delay after a keypress before applying the filter. This practice, known as “debouncing” user input, involves disregarding rapid events and focusing on more stable ones.

When using the debounce() operator, you specify a time period in milliseconds. This operator receives input from the upstream flow and only forwards it downstream if there has been no new input within the specified time period. Additionally, when the upstream flow is closed, the debounce() operator emits the last result it received.

Let’s go though a small piece of code.

• The debounce() operator has a delay of 50 milliseconds.
• 0 is emitted by the upstream flow and given to debounce().
• After 100 milliseconds, 1 is emitted by the upstream flow.
• debounce() discards the 0 result because it was replaced too quickly.
• After 200 milliseconds, 2 is emitted by the upstream flow.
• debounce() discards the 1 result for the same reason.
• After 50 milliseconds, 3 is emitted by the upstream flow.
• debounce() discards the 2 result.
• Now delay is 400 milliseconds, so debounce() emits 3 since no new result was received within 250 milliseconds.
• 4 is emitted by the upstream flow.
• The upstream flow is closed.
• debounce() emits 4 since the upstream flow was closed.

In simpler terms, debounce() waits for a specific time period before considering a result as valid. It discards results that are replaced too quickly and emits the last result if the upstream flow is closed.

Thus final output will be
3
4

drop(): Skipping the Start

In certain cases, you may want to discard the initial items emitted by a flow. This can occur when you are aware that the first few items are part of a “warmup” phase and the actual meaningful data starts later.

the first four count of item will be dropped and the out will be

5, 6, 7, 8, 9, 10

dropWhile(): Skip Until You’re Ready

The dropWhile() function skips all items in a flow until a provided lambda expression or function returns true. Once that condition is met, the item that triggered the condition and all subsequent items are emitted.

In this scenario, using dropWhile(), we discard items from the flow as long as they are odd. However, as soon as we encounter the first even item, we stop dropping and start emitting all the remaining items as they come.

2, 3, 4, 5, 6, 7, 8, 9, 10

filter(): Take What You Want

The filter() function evaluates a provided lambda expression or function on each item emitted by the upstream flow. If the expression evaluates to true for an item, it is passed downstream and emitted.

3, 6, 9

filterIsInstance(): Type Safety, Flow Style

By using filterIsInstance(), you can selectively filter emissions from the upstream flow based on their type. Only items that match the specified type will be allowed to pass through and be emitted downstream.

Only strings will be filtered out.

note: filterIsInstance() helps when you have a Flow with a common supertype, but you specifically want a Flow of a particular sub-type.

filterNot(): Take The Other Stuff Instead

The filterNot() function is similar to filter(), but with a negation (!) applied. It only emits items from the upstream flow where the provided lambda expression evaluates to false, whereas filter() looks for items where the expression evaluates to true. In other words, filterNot() selects items that do not satisfy the given condition, while filter() selects items that do satisfy the condition.

1, 2, 4, 5, 7, 8, 10

For more details you can go though book Elements of Kotlin coroutine

Happy Learning!
Please share any suggestion.