In this section, we'll explore some essential operations you can perform on streams. These operations allow you to manipulate and interact with stream elements in various ways.

#Tapping

Tapping is an operation that involves running an effect on each emission of the stream. It allows you to observe each element, perform some effectful operation, and discard the result of this observation. Importantly, the Stream.tap operation does not alter the elements of the stream, and it does not affect the return type of the stream.

For instance, you can use Stream.tap to print each element of a stream:

ts
import { Stream, Console, Effect } from "effect"
 
const stream = Stream.make(1, 2, 3).pipe(
  Stream.tap((n) => Console.log(`before mapping: ${n}`)),
  Stream.map((n) => n * 2),
  Stream.tap((n) => Console.log(`after mapping: ${n}`))
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
before mapping: 1
after mapping: 2
before mapping: 2
after mapping: 4
before mapping: 3
after mapping: 6
{
  _id: "Chunk",
  values: [ 2, 4, 6 ]
}
*/

ts
import { Stream, Console, Effect } from "effect"
 
const stream = Stream.make(1, 2, 3).pipe(
  Stream.tap((n) => Console.log(`before mapping: ${n}`)),
  Stream.map((n) => n * 2),
  Stream.tap((n) => Console.log(`after mapping: ${n}`))
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
before mapping: 1
after mapping: 2
before mapping: 2
after mapping: 4
before mapping: 3
after mapping: 6
{
  _id: "Chunk",
  values: [ 2, 4, 6 ]
}
*/

#Taking Elements

Another essential operation is taking elements, which allows you to extract a specific number of elements from a stream. Here are several ways to achieve this:

• take. To extract a fixed number of elements.
• takeWhile. To extract elements until a certain condition is met.
• takeUntil. To extract elements until a specific condition is met.
• takeRight. To extract a specified number of elements from the end.

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.iterate(0, (n) => n + 1)
 
// Using `take` to extract a fixed number of elements:
const s1 = Stream.take(stream, 5)
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4 ]
}
*/
 
// Using `takeWhile` to extract elements until a certain condition is met:
const s2 = Stream.takeWhile(stream, (n) => n < 5)
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4 ]
}
*/
 
// Using `takeUntil` to extract elements until a specific condition is met:
const s3 = Stream.takeUntil(stream, (n) => n === 5)
Effect.runPromise(Stream.runCollect(s3)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4, 5 ]
}
*/
 
// Using `takeRight` to extract a specified number of elements from the end:
const s4 = Stream.takeRight(s3, 3)
Effect.runPromise(Stream.runCollect(s4)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 3, 4, 5 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.iterate(0, (n) => n + 1)
 
// Using `take` to extract a fixed number of elements:
const s1 = Stream.take(stream, 5)
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4 ]
}
*/
 
// Using `takeWhile` to extract elements until a certain condition is met:
const s2 = Stream.takeWhile(stream, (n) => n < 5)
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4 ]
}
*/
 
// Using `takeUntil` to extract elements until a specific condition is met:
const s3 = Stream.takeUntil(stream, (n) => n === 5)
Effect.runPromise(Stream.runCollect(s3)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 2, 3, 4, 5 ]
}
*/
 
// Using `takeRight` to extract a specified number of elements from the end:
const s4 = Stream.takeRight(s3, 3)
Effect.runPromise(Stream.runCollect(s4)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 3, 4, 5 ]
}
*/

#Exploring Streams as an Alternative to Async Iterables

When working with asynchronous data sources, like async iterables, you often need to consume data in a loop until a certain condition is met. This tutorial introduces how you can achieve similar behavior using Streams in a beginner-friendly manner.

In async iterables, data consumption can continue in a loop until a break or return statement is encountered. To replicate this behavior with Streams, you have a couple of options:

1. Stream.takeUntil: This function allows you to take elements from a stream until a specified condition evaluates to true. It's akin to breaking out of a loop in async iterables when a certain condition is met.

2. Stream.toPull: The Stream.toPull function is another way to replicate looping through async iterables. It returns an effect that repeatedly pulls data chunks from the stream. This effect can fail with None when the stream is finished or with Some error if it fails.

Let's take a closer look at the second option, Stream.toPull.

ts
import { Stream, Effect } from "effect"
 
// Simulate a chunked stream
const stream = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(Stream.rechunk(2))
 
const program = Effect.gen(function* (_) {
  // Create an effect to get data chunks from the stream
  const getChunk = yield* _(Stream.toPull(stream))
 
  // Continuously fetch and process chunks
  while (true) {
    const chunk = yield* _(getChunk)
    console.log(chunk)
  }
})
 
Effect.runPromise(Effect.scoped(program)).then(console.log, console.error)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2 ] }
{ _id: 'Chunk', values: [ 3, 4 ] }
{ _id: 'Chunk', values: [ 5 ] }
{
  _id: 'FiberFailure',
  cause: {
    _id: 'Cause',
    _tag: 'Fail',
    failure: { _id: 'Option', _tag: 'None' }
  }
}
*/

ts
import { Stream, Effect } from "effect"
 
// Simulate a chunked stream
const stream = Stream.fromIterable([1, 2, 3, 4, 5]).pipe(Stream.rechunk(2))
 
const program = Effect.gen(function* (_) {
  // Create an effect to get data chunks from the stream
  const getChunk = yield* _(Stream.toPull(stream))
 
  // Continuously fetch and process chunks
  while (true) {
    const chunk = yield* _(getChunk)
    console.log(chunk)
  }
})
 
Effect.runPromise(Effect.scoped(program)).then(console.log, console.error)
/*
Output:
{ _id: 'Chunk', values: [ 1, 2 ] }
{ _id: 'Chunk', values: [ 3, 4 ] }
{ _id: 'Chunk', values: [ 5 ] }
{
  _id: 'FiberFailure',
  cause: {
    _id: 'Cause',
    _tag: 'Fail',
    failure: { _id: 'Option', _tag: 'None' }
  }
}
*/

In this example, we're using Stream.toPull to repeatedly pull data chunks from the stream. The code enters a loop and continues to fetch and display chunks until there's no more data left to process.

#Mapping

In this section, we'll explore how to transform elements within a stream using the Stream.map family of operations. These operations allow you to apply a function to each element of the stream, producing a new stream with the transformed values.

#Basic Mapping

The Stream.map operation applies a given function to all elements of the stream, creating another stream with the transformed values. Let's illustrate this with an example:

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make(1, 2, 3).pipe(Stream.map((n) => n + 1))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 3, 4 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make(1, 2, 3).pipe(Stream.map((n) => n + 1))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 3, 4 ]
}
*/

#Effectful Mapping

If your transformation involves effects, you can use Stream.mapEffect instead. It allows you to apply an effectful function to each element of the stream, producing a new stream with effectful results:

ts
import { Stream, Random, Effect } from "effect"
 
const stream = Stream.make(10, 20, 30).pipe(
  Stream.mapEffect((n) => Random.nextIntBetween(0, n))
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 6, 13, 5 ]
}
*/

ts
import { Stream, Random, Effect } from "effect"
 
const stream = Stream.make(10, 20, 30).pipe(
  Stream.mapEffect((n) => Random.nextIntBetween(0, n))
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 6, 13, 5 ]
}
*/

You can evaluate effects concurrently using the concurrency option. It allows you to specify the number of concurrent running effects. The results are emitted downstream in the original order.

Let's write a simple page downloader that fetches URLs concurrently:

ts
import { Stream, Effect } from "effect"
 
const getUrls = Effect.succeed(["url0", "url1", "url2"])
 
const fetchUrl = (url: string) =>
  Effect.succeed([
    `Resource 0-${url}`,
    `Resource 1-${url}`,
    `Resource 2-${url}`
  ])
 
const stream = Stream.fromIterableEffect(getUrls).pipe(
  Stream.mapEffect(fetchUrl, { concurrency: 4 })
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "Resource 0-url0", "Resource 1-url0", "Resource 2-url0" ], [ "Resource 0-url1", "Resource 1-url1",
      "Resource 2-url1" ], [ "Resource 0-url2", "Resource 1-url2", "Resource 2-url2" ]
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const getUrls = Effect.succeed(["url0", "url1", "url2"])
 
const fetchUrl = (url: string) =>
  Effect.succeed([
    `Resource 0-${url}`,
    `Resource 1-${url}`,
    `Resource 2-${url}`
  ])
 
const stream = Stream.fromIterableEffect(getUrls).pipe(
  Stream.mapEffect(fetchUrl, { concurrency: 4 })
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "Resource 0-url0", "Resource 1-url0", "Resource 2-url0" ], [ "Resource 0-url1", "Resource 1-url1",
      "Resource 2-url1" ], [ "Resource 0-url2", "Resource 1-url2", "Resource 2-url2" ]
  ]
}
*/

#Stateful Mapping

The Stream.mapAccum operation is similar to map, but it transforms elements statefully and allows you to map and accumulate in a single operation. Let's see how you can use it to calculate the running total of an input stream:

ts
import { Stream, Effect } from "effect"
 
const runningTotal = (stream: Stream.Stream<number>): Stream.Stream<number> =>
  stream.pipe(Stream.mapAccum(0, (s, a) => [s + a, s + a]))
 
// input:  0, 1, 2, 3, 4, 5
Effect.runPromise(Stream.runCollect(runningTotal(Stream.range(0, 6)))).then(
  console.log
)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 3, 6, 10, 15 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const runningTotal = (stream: Stream.Stream<number>): Stream.Stream<number> =>
  stream.pipe(Stream.mapAccum(0, (s, a) => [s + a, s + a]))
 
// input:  0, 1, 2, 3, 4, 5
Effect.runPromise(Stream.runCollect(runningTotal(Stream.range(0, 6)))).then(
  console.log
)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 3, 6, 10, 15 ]
}
*/

#Mapping and Flattening

The Stream.mapConcat operation is akin to Stream.map, but it takes things a step further. It maps each element to zero or more elements of type Iterable and then flattens the entire stream. Let's illustrate this with an example:

ts
import { Stream, Effect } from "effect"
 
const numbers = Stream.make("1-2-3", "4-5", "6").pipe(
  Stream.mapConcat((s) => s.split("-")),
  Stream.map((s) => parseInt(s))
)
 
Effect.runPromise(Stream.runCollect(numbers)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 1, 2, 3, 4, 5, 6 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const numbers = Stream.make("1-2-3", "4-5", "6").pipe(
  Stream.mapConcat((s) => s.split("-")),
  Stream.map((s) => parseInt(s))
)
 
Effect.runPromise(Stream.runCollect(numbers)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 1, 2, 3, 4, 5, 6 ]
}
*/

In this example, we take a stream of strings like "1-2-3" and split them into individual numbers, resulting in a flattened stream of integers.

#Mapping to a Constant Value

The Stream.as method allows you to map the success values of a stream to a specified constant value. This can be handy when you want to transform elements into a uniform value. Here's an example where we map all elements to the null value:

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 5).pipe(Stream.as(null))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ null, null, null, null ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 5).pipe(Stream.as(null))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ null, null, null, null ]
}
*/

In this case, regardless of the original values in the stream, we've mapped them all to null.

#Filtering

The Stream.filter operation is like a sieve that lets through elements that meet a specified condition. Think of it as a way to sift through a stream and keep only the elements that satisfy the given criteria. Here's an example:

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 11).pipe(Stream.filter((n) => n % 2 === 0))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8, 10 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 11).pipe(Stream.filter((n) => n % 2 === 0))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8, 10 ]
}
*/

In this example, we start with a stream of numbers from 1 to 10 and use Stream.filter to retain only the even numbers (those that satisfy the condition n % 2 === 0). The result is a filtered stream containing the even numbers from the original stream.

#Scanning

In this section, we'll explore the concept of stream scanning. Scans are similar to folds, but they provide a historical perspective. Like folds, scans also involve a binary operator and an initial value. However, what makes scans unique is that they emit every intermediate result as part of the stream.

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 6).pipe(Stream.scan(0, (a, b) => a + b))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 3, 6, 10, 15 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(1, 6).pipe(Stream.scan(0, (a, b) => a + b))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 0, 1, 3, 6, 10, 15 ]
}
*/

In this example, we have a stream of numbers from 1 to 5, and we use Stream.scan to perform a cumulative addition starting from an initial value of 0. The result is a stream that emits the accumulated sum at each step: 0, 1, 3, 6, 10, and 15.

Streams scans provide a way to keep a historical record of your stream transformations, which can be invaluable for various applications.

Additionally, if you only need the final result of the scan, you can use Stream.runFold:

ts
import { Stream, Effect } from "effect"
 
const fold = Stream.range(1, 6).pipe(Stream.runFold(0, (a, b) => a + b))
 
Effect.runPromise(fold).then(console.log) // Output: 15

ts
import { Stream, Effect } from "effect"
 
const fold = Stream.range(1, 6).pipe(Stream.runFold(0, (a, b) => a + b))
 
Effect.runPromise(fold).then(console.log) // Output: 15

In this case, Stream.runFold gives you the final accumulated value, which is 15 in this example.

#Draining

In this section, we'll explore the concept of stream draining. Imagine you have a stream filled with effectful operations, but you're not interested in the values they produce; instead, you want to execute these effects and discard the results. This is where the Stream.drain function comes into play.

Let's go through a few examples:

Example 1: Discarding Values

ts
import { Stream, Effect } from "effect"
 
// We create a stream and immediately drain it.
const s1 = Stream.range(1, 6).pipe(Stream.drain)
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: []
}
*/

ts
import { Stream, Effect } from "effect"
 
// We create a stream and immediately drain it.
const s1 = Stream.range(1, 6).pipe(Stream.drain)
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: []
}
*/

In this example, we have a stream with values from 1 to 5, but we use Stream.drain to discard these values. As a result, the output stream is empty.

Example 2: Executing Random Effects

ts
import { Stream, Effect, Random } from "effect"
 
const s2 = Stream.repeatEffect(
  Effect.gen(function* (_) {
    const nextInt = yield* _(Random.nextInt)
    const number = Math.abs(nextInt % 10)
    console.log(`random number: ${number}`)
    return number
  })
).pipe(Stream.take(3))
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
random number: 4
random number: 2
random number: 7
{
  _id: "Chunk",
  values: [ 4, 2, 7 ]
}
*/
 
const s3 = Stream.drain(s2)
 
Effect.runPromise(Stream.runCollect(s3)).then(console.log)
/*
random number: 1
random number: 6
random number: 0
Output:
{
  _id: "Chunk",
  values: []
}
*/

ts
import { Stream, Effect, Random } from "effect"
 
const s2 = Stream.repeatEffect(
  Effect.gen(function* (_) {
    const nextInt = yield* _(Random.nextInt)
    const number = Math.abs(nextInt % 10)
    console.log(`random number: ${number}`)
    return number
  })
).pipe(Stream.take(3))
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
random number: 4
random number: 2
random number: 7
{
  _id: "Chunk",
  values: [ 4, 2, 7 ]
}
*/
 
const s3 = Stream.drain(s2)
 
Effect.runPromise(Stream.runCollect(s3)).then(console.log)
/*
random number: 1
random number: 6
random number: 0
Output:
{
  _id: "Chunk",
  values: []
}
*/

In this example, we create a stream with random effects and collect the values of these effects initially. Later, we use Stream.drain to execute the same effects without collecting the values. This demonstrates how you can use draining to trigger effectful operations when you're not interested in the emitted values.

Stream draining can be particularly useful when you need to perform certain actions or cleanup tasks in your application without affecting the main stream of data.

#Detecting Changes in a Stream

In this section, we'll explore the Stream.changes operation, which allows you to detect and emit elements that are different from their preceding elements within the stream.

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make(1, 1, 1, 2, 2, 3, 4).pipe(Stream.changes)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 1, 2, 3, 4 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make(1, 1, 1, 2, 2, 3, 4).pipe(Stream.changes)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [ 1, 2, 3, 4 ]
}
*/

#Zipping

Zipping is a process of combining two or more streams to create a new stream by pairing elements from the input streams. We can achieve this using the Stream.zip and Stream.zipWith operators. Let's dive into some examples:

ts
import { Stream, Effect } from "effect"
 
// We create two streams and zip them together.
const s1 = Stream.zip(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c")
)
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 2, "b" ], [ 3, "c" ]
  ]
}
*/
 
// We create two streams and zip them with custom logic.
const s2 = Stream.zipWith(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c"),
  (n, s) => [n - s.length, s]
)
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 0, "a" ], [ 1, "b" ], [ 2, "c" ]
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
// We create two streams and zip them together.
const s1 = Stream.zip(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c")
)
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 2, "b" ], [ 3, "c" ]
  ]
}
*/
 
// We create two streams and zip them with custom logic.
const s2 = Stream.zipWith(
  Stream.make(1, 2, 3, 4, 5, 6),
  Stream.make("a", "b", "c"),
  (n, s) => [n - s.length, s]
)
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 0, "a" ], [ 1, "b" ], [ 2, "c" ]
  ]
}
*/

The new stream will end when one of the streams ends.

#Handling Stream Endings

When one of the input streams ends before the other, you might need to zip with default values. The Stream.zipAll and Stream.zipAllWith operations allow you to specify default values for both sides to handle such scenarios. Let's see an example:

ts
import { Stream, Effect } from "effect"
 
const s1 = Stream.zipAll(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  defaultSelf: 0,
  defaultOther: "x"
})
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 2, "b" ], [ 3, "c" ], [ 4, "x" ], [ 5, "x" ], [ 6, "x" ]
  ]
}
*/
 
const s2 = Stream.zipAllWith(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  onSelf: (n) => [n, "x"],
  onOther: (s) => [0, s],
  onBoth: (n, s) => [n - s.length, s]
})
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 0, "a" ], [ 1, "b" ], [ 2, "c" ], [ 4, "x" ], [ 5, "x" ], [ 6, "x" ]
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const s1 = Stream.zipAll(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  defaultSelf: 0,
  defaultOther: "x"
})
 
Effect.runPromise(Stream.runCollect(s1)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 2, "b" ], [ 3, "c" ], [ 4, "x" ], [ 5, "x" ], [ 6, "x" ]
  ]
}
*/
 
const s2 = Stream.zipAllWith(Stream.make(1, 2, 3, 4, 5, 6), {
  other: Stream.make("a", "b", "c"),
  onSelf: (n) => [n, "x"],
  onOther: (s) => [0, s],
  onBoth: (n, s) => [n - s.length, s]
})
 
Effect.runPromise(Stream.runCollect(s2)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 0, "a" ], [ 1, "b" ], [ 2, "c" ], [ 4, "x" ], [ 5, "x" ], [ 6, "x" ]
  ]
}
*/

This allows you to handle zipping when one stream completes earlier than the other.

#Zipping Streams at Different Rates

Sometimes, you might have two streams producing elements at different speeds. If you don't want to wait for the slower one when zipping elements, you can use Stream.zipLatest or Stream.zipLatestWith. These operations combine elements in a way that when a value is emitted by either of the two streams, it is combined with the latest value from the other stream to produce a result. Here's an example:

ts
import { Stream, Schedule, Effect } from "effect"
 
const s1 = Stream.make(1, 2, 3).pipe(
  Stream.schedule(Schedule.spaced("1 seconds"))
)
 
const s2 = Stream.make("a", "b", "c", "d").pipe(
  Stream.schedule(Schedule.spaced("500 millis"))
)
 
const stream = Stream.zipLatest(s1, s2)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 1, "b" ], [ 2, "b" ], [ 2, "c" ], [ 2, "d" ], [ 3, "d" ]
  ]
}
*/

ts
import { Stream, Schedule, Effect } from "effect"
 
const s1 = Stream.make(1, 2, 3).pipe(
  Stream.schedule(Schedule.spaced("1 seconds"))
)
 
const s2 = Stream.make("a", "b", "c", "d").pipe(
  Stream.schedule(Schedule.spaced("500 millis"))
)
 
const stream = Stream.zipLatest(s1, s2)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 1, "b" ], [ 2, "b" ], [ 2, "c" ], [ 2, "d" ], [ 3, "d" ]
  ]
}
*/

Here, Stream.zipLatest combines elements from both streams without waiting for the slower one, resulting in a more responsive output.

#Pairing with Previous and Next Elements

• zipWithPrevious: This operator pairs each element of a stream with its previous element.

• zipWithNext: It pairs each element of a stream with its next element.

• zipWithPreviousAndNext: This operator pairs each element with both its previous and next elements.

Here's an example illustrating these operations:

ts
import { Stream } from "effect"
 
const stream = Stream.make(1, 2, 3, 4)
 
const s1 = Stream.zipWithPrevious(stream)
 
const s2 = Stream.zipWithNext(stream)
 
const s3 = Stream.zipWithPreviousAndNext(stream)

ts
import { Stream } from "effect"
 
const stream = Stream.make(1, 2, 3, 4)
 
const s1 = Stream.zipWithPrevious(stream)
 
const s2 = Stream.zipWithNext(stream)
 
const s3 = Stream.zipWithPreviousAndNext(stream)

#Indexing Stream Elements

Another handy operator is Stream.zipWithIndex, which indexes each element of a stream by pairing it with its respective index. This is especially useful when you need to keep track of the position of elements within the stream.

Here's an example of indexing elements in a stream:

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make("Mary", "James", "Robert", "Patricia")
 
const indexedStream = Stream.zipWithIndex(stream)
 
Effect.runPromise(Stream.runCollect(indexedStream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "Mary", 0 ], [ "James", 1 ], [ "Robert", 2 ], [ "Patricia", 3 ]
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.make("Mary", "James", "Robert", "Patricia")
 
const indexedStream = Stream.zipWithIndex(stream)
 
Effect.runPromise(Stream.runCollect(indexedStream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "Mary", 0 ], [ "James", 1 ], [ "Robert", 2 ], [ "Patricia", 3 ]
  ]
}
*/

#Cartesian Product of Streams

The Stream module introduces a powerful feature: the ability to compute the Cartesian Product of two streams. This operation allows you to generate combinations of elements from two separate streams. Let's explore this concept further:

Imagine you have two sets of items, and you want to generate all possible pairs by taking one item from each set. This process is known as finding the Cartesian Product of the sets. In the context of streams, it means creating combinations of elements from two streams.

To achieve this, the Stream module provides the Stream.cross operator, along with its variants. These operators take two streams and generate a new stream containing all possible combinations of elements from the original streams.

Here's a practical example:

ts
import { Stream, Effect } from "effect"
 
const s1 = Stream.make(1, 2, 3)
const s2 = Stream.make("a", "b")
 
const product = Stream.cross(s1, s2)
 
Effect.runPromise(Stream.runCollect(product)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 1, "b" ], [ 2, "a" ], [ 2, "b" ], [ 3, "a" ], [ 3, "b" ]
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const s1 = Stream.make(1, 2, 3)
const s2 = Stream.make("a", "b")
 
const product = Stream.cross(s1, s2)
 
Effect.runPromise(Stream.runCollect(product)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 1, "a" ], [ 1, "b" ], [ 2, "a" ], [ 2, "b" ], [ 3, "a" ], [ 3, "b" ]
  ]
}
*/

It's important to note that the right-hand side stream (s2 in this case) will be iterated multiple times, once for each element in the left-hand side stream (s1). This means that if the right-hand side stream involves expensive or side-effectful operations, they will be executed multiple times.

#Partitioning

Partitioning a stream means dividing it into two separate streams based on a specified condition. The Stream module provides two helpful functions for achieving this: Stream.partition and Stream.partitionEither. Let's explore how these functions work and when to use them.

#partition

The Stream.partition function takes a predicate as input and splits the original stream into two substreams: one containing elements that satisfy the predicate (evaluate to true), and the other containing elements that do not (evaluate to false). Additionally, these substreams are wrapped within a Scope type.

Here's an example where we partition a stream of numbers into even and odd numbers:

ts
import { Stream, Effect } from "effect"
 
const partition = Stream.range(1, 10).pipe(
  Stream.partition((n) => n % 2 === 0, { bufferSize: 5 })
)
 
Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* (_) {
      const [evens, odds] = yield* _(partition)
      console.log(yield* _(Stream.runCollect(evens)))
      console.log(yield* _(Stream.runCollect(odds)))
    })
  )
)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8 ]
}
{
  _id: "Chunk",
  values: [ 1, 3, 5, 7, 9 ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const partition = Stream.range(1, 10).pipe(
  Stream.partition((n) => n % 2 === 0, { bufferSize: 5 })
)
 
Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* (_) {
      const [evens, odds] = yield* _(partition)
      console.log(yield* _(Stream.runCollect(evens)))
      console.log(yield* _(Stream.runCollect(odds)))
    })
  )
)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8 ]
}
{
  _id: "Chunk",
  values: [ 1, 3, 5, 7, 9 ]
}
*/

In this example, we use the Stream.partition function with a predicate to split the stream into even and odd numbers. The bufferSize option controls how much the faster stream can advance beyond the slower one.

#partitionEither

Sometimes, you may need to partition a stream using an effectful predicate. For such cases, the Stream.partitionEither function is available. This function accepts an effectful predicate and divides the stream into two substreams based on the result of the predicate: elements that yield Either.left values go to one substream, while elements yielding Either.right values go to the other.

Here's an example where we use Stream.partitionEither to partition a stream of numbers based on an effectful condition:

ts
import { Stream, Effect, Either } from "effect"
 
const partition = Stream.range(1, 10).pipe(
  Stream.partitionEither(
    (n) => Effect.succeed(n < 5 ? Either.left(n * 2) : Either.right(n)),
    { bufferSize: 5 }
  )
)
 
Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* (_) {
      const [evens, odds] = yield* _(partition)
      console.log(yield* _(Stream.runCollect(evens)))
      console.log(yield* _(Stream.runCollect(odds)))
    })
  )
)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8 ]
}
{
  _id: "Chunk",
  values: [ 5, 6, 7, 8, 9 ]
}
*/

ts
import { Stream, Effect, Either } from "effect"
 
const partition = Stream.range(1, 10).pipe(
  Stream.partitionEither(
    (n) => Effect.succeed(n < 5 ? Either.left(n * 2) : Either.right(n)),
    { bufferSize: 5 }
  )
)
 
Effect.runPromise(
  Effect.scoped(
    Effect.gen(function* (_) {
      const [evens, odds] = yield* _(partition)
      console.log(yield* _(Stream.runCollect(evens)))
      console.log(yield* _(Stream.runCollect(odds)))
    })
  )
)
/*
Output:
{
  _id: "Chunk",
  values: [ 2, 4, 6, 8 ]
}
{
  _id: "Chunk",
  values: [ 5, 6, 7, 8, 9 ]
}
*/

In this case, the Stream.partitionEither function splits the stream into two substreams: one containing values that are less than 5 (doubled using Either.left), and the other containing values greater than or equal to 5 (using Either.right).

#GroupBy

When working with streams of data, you may often need to group elements based on certain criteria. The Stream module provides two functions for achieving this: groupByKey and groupBy. Let's explore how these functions work and when to use them.

#groupByKey

The Stream.groupByKey function allows you to partition a stream by a simple function of type (a: A) => K, where A represents the type of elements in your stream, and K represents the keys by which the stream should be partitioned. This function is not effectful, it simply groups elements by applying the provided function.

The Stream.groupByKey function returns a new data type called GroupBy. This GroupBy type represents a grouped stream. To work with the groups, you can use the GroupBy.evaluate function, which takes a function of type (key: K, stream: Stream<V, E>) => Stream.Stream<...>. This function runs across all groups and merges them in a non-deterministic fashion.

In the example below, we use groupByKey to group exam results by the tens place of their scores and count the number of results in each group:

ts
import { Stream, GroupBy, Effect, Chunk } from "effect"
 
class Exam {
  constructor(
    readonly person: string,
    readonly score: number
  ) {}
}
 
const examResults = [
  new Exam("Alex", 64),
  new Exam("Michael", 97),
  new Exam("Bill", 77),
  new Exam("John", 78),
  new Exam("Bobby", 71)
]
 
const groupByKeyResult = Stream.fromIterable(examResults).pipe(
  Stream.groupByKey((exam) => Math.floor(exam.score / 10) * 10)
)
 
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.map((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 60, 1 ], [ 90, 1 ], [ 70, 3 ]
  ]
}
*/

ts
import { Stream, GroupBy, Effect, Chunk } from "effect"
 
class Exam {
  constructor(
    readonly person: string,
    readonly score: number
  ) {}
}
 
const examResults = [
  new Exam("Alex", 64),
  new Exam("Michael", 97),
  new Exam("Bill", 77),
  new Exam("John", 78),
  new Exam("Bobby", 71)
]
 
const groupByKeyResult = Stream.fromIterable(examResults).pipe(
  Stream.groupByKey((exam) => Math.floor(exam.score / 10) * 10)
)
 
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.map((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ 60, 1 ], [ 90, 1 ], [ 70, 3 ]
  ]
}
*/

In this example, we partition the exam results into groups based on the tens place of their scores (e.g., 60, 90, 70). The groupByKey function is ideal for simple, non-effectful partitioning.

#groupBy

In more complex scenarios where partitioning involves effects, you can turn to the Stream.groupBy function. This function takes an effectful partitioning function and generates a GroupBy data type, representing a grouped stream. You can then use GroupBy.evaluate in a similar fashion as before to process the groups.

In the following example, we group names by their first character and count the number of names in each group. Note that the partitioning operation itself is simulated as effectful:

ts
import { Stream, GroupBy, Effect, Chunk } from "effect"
 
const groupByKeyResult = Stream.fromIterable([
  "Mary",
  "James",
  "Robert",
  "Patricia",
  "John",
  "Jennifer",
  "Rebecca",
  "Peter"
]).pipe(
  Stream.groupBy((name) => Effect.succeed([name.substring(0, 1), name]))
)
 
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.map((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "M", 1 ], [ "J", 3 ], [ "R", 2 ], [ "P", 2 ]
  ]
}
*/

ts
import { Stream, GroupBy, Effect, Chunk } from "effect"
 
const groupByKeyResult = Stream.fromIterable([
  "Mary",
  "James",
  "Robert",
  "Patricia",
  "John",
  "Jennifer",
  "Rebecca",
  "Peter"
]).pipe(
  Stream.groupBy((name) => Effect.succeed([name.substring(0, 1), name]))
)
 
const stream = GroupBy.evaluate(groupByKeyResult, (key, stream) =>
  Stream.fromEffect(
    Stream.runCollect(stream).pipe(
      Effect.map((chunk) => [key, Chunk.size(chunk)] as const)
    )
  )
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    [ "M", 1 ], [ "J", 3 ], [ "R", 2 ], [ "P", 2 ]
  ]
}
*/

#Grouping

When working with streams, you may encounter situations where you need to group elements in a more structured manner. The Stream module provides two helpful functions for achieving this: grouped and groupedWithin. In this section, we'll explore how these functions work and when to use them.

#grouped

The Stream.grouped function is perfect for partitioning stream results into chunks of a specified size. It's especially useful when you want to work with data in smaller, more manageable pieces.

Here's an example that demonstrates the use of Stream.grouped:

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(0, 8).pipe(Stream.grouped(3))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    {
      _id: "Chunk",
      values: [ 0, 1, 2 ]
    }, {
      _id: "Chunk",
      values: [ 3, 4, 5 ]
    }, {
      _id: "Chunk",
      values: [ 6, 7 ]
    }
  ]
}
*/

ts
import { Stream, Effect } from "effect"
 
const stream = Stream.range(0, 8).pipe(Stream.grouped(3))
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    {
      _id: "Chunk",
      values: [ 0, 1, 2 ]
    }, {
      _id: "Chunk",
      values: [ 3, 4, 5 ]
    }, {
      _id: "Chunk",
      values: [ 6, 7 ]
    }
  ]
}
*/

In this example, we take a stream of numbers from 0 to 9 and use Stream.grouped(3) to divide it into chunks of size 3.

#groupedWithin

The Stream.groupedWithin function provides more flexibility by allowing you to group events based on time intervals or chunk size, whichever condition is satisfied first. This is particularly useful when you want to group data based on time constraints.

ts
import { Stream, Schedule, Effect } from "effect"
 
const stream = Stream.range(0, 10).pipe(
  Stream.repeat(Schedule.spaced("1 seconds")),
  Stream.groupedWithin(18, "1.5 seconds"),
  Stream.take(3)
)
 
Effect.runPromise(Stream.runCollect(stream)).then(console.log)
/*
Output:
{
  _id: "Chunk",
  values: [
    {
      _id: "Chunk",
      values: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7 ]
    }, {
      _id: "Chunk",
      values: [ 8, 9, 0, 1, 2, 3, 4, 5, 6, 7, 8, 9 ]
    }, {
      _id: "Chunk",
      values: [ 0, 1, 2, 3, 4, 5, 6, 7, 8, 9, 0, 1, 2, 3, 4, 5, 6, 7 ]
    }
  ]
}
*/