Channel

import { Channel, Effect } from "effect"

// Create two channels
const leftChannel = Channel.fromIterable([1, 2, 3])
const rightChannel = Channel.fromIterable(["a", "b", "c"])

// The default "both" strategy waits for both channels to complete
const mergedChannel = Channel.merge(leftChannel, rightChannel)

const values = await Effect.runPromise(Channel.runCollect(mergedChannel))
values.map(String).sort() // => ["1", "2", "3", "a", "b", "c"]

import { Channel, Data, Effect } from "effect"

class MergeAllError extends Data.TaggedError("MergeAllError")<{
  readonly reason: string
}> {}

// Create channels that output other channels
const nestedChannels = Channel.fromIterable([
  Channel.fromIterable([1, 2]),
  Channel.fromIterable([3, 4]),
  Channel.fromIterable([5, 6])
])

// Merge all channels with bounded concurrency
const mergedChannel = Channel.mergeAll({
  concurrency: 1,
  bufferSize: 16
})(nestedChannels)

await Effect.runPromise(Channel.runCollect(mergedChannel)) // => [1, 2, 3, 4, 5, 6]

import { Channel } from "effect"

Channel.DefaultChunkSize // => 4096

import { Channel, Effect } from "effect"

const released: Array<string> = []
const channel = Channel.acquireRelease(
  Effect.succeed("resource"),
  (resource, exit) => Effect.sync(() => released.push(resource))
)
const observed = [await Effect.runPromise(Channel.runCollect(channel)), released] // => [["resource"], ["resource"]]

import { Channel, Effect } from "effect"

const released: Array<string> = []
const channel = Channel.acquireUseRelease(
  Effect.succeed("resource"),
  (resource) => Channel.succeed(resource.toUpperCase()),
  (resource, exit) => Effect.sync(() => released.push(resource))
)
const observed = [await Effect.runPromise(Channel.runCollect(channel)), released] // => [["RESOURCE"], ["resource"]]

import { Channel, Effect, Queue } from "effect"

const channel = Channel.callback<number>((queue) =>
  Effect.gen(function*() {
    yield* Queue.offer(queue, 1)
    yield* Queue.offer(queue, 2)
    yield* Queue.offer(queue, 3)
    yield* Queue.end(queue)
  })
)
await Effect.runPromise(Channel.runCollect(channel)) // => [1, 2, 3]

import { Channel, Effect, Queue } from "effect"

const channel = Channel.callbackArray<number>(Effect.fn(function*(queue) {
  yield* Queue.offer(queue, 1)
  yield* Queue.offer(queue, 2)
  yield* Queue.end(queue)
}))
await Effect.runPromise(Channel.runCollect(channel)) // => [[1, 2]]

import { Cause, Channel, Effect, Exit } from "effect"

const defect = "Unrecoverable error"
const diedChannel = Channel.die(defect)
Effect.runSync(Effect.exit(Channel.runCollect(diedChannel))) // => Exit.failCause(Cause.die(defect))

import { Channel, Effect } from "effect"

// Create a channel that outputs values
const sourceChannel = Channel.fromIterable([1, 2, 3, 4, 5])

// Drain all output, keeping only the completion
const drainedChannel = Channel.drain(sourceChannel)

Effect.runSync(Channel.runCollect(drainedChannel)) // => []

import { Channel, Effect } from "effect"

// Create an empty channel
const emptyChannel = Channel.empty

// Use empty channel in composition
const combined = Channel.concatWith(emptyChannel, () => Channel.succeed(42))
// Will immediately provide the second channel's output

// Empty channel can be used as a no-op in conditional logic
const conditionalChannel = (shouldEmit: boolean) =>
  shouldEmit ? Channel.succeed("data") : Channel.empty

Effect.runSync(Channel.runCollect(conditionalChannel(true))) // => ["data"]

import { Channel, Effect } from "effect"

const channel = Channel.end("done")
Effect.runSync(Channel.runCollect(channel)) // => []

import { Channel, Effect, Exit } from "effect"

const failedChannel = Channel.fail("Something went wrong")
Effect.runSync(Effect.exit(Channel.runCollect(failedChannel))) // => Exit.fail("Something went wrong")

import { Cause, Channel, Effect, Exit } from "effect"

const simpleCause = Cause.fail("Simple error")
const failedChannel = Channel.failCause(simpleCause)
Effect.runSync(Effect.exit(Channel.runCollect(failedChannel))) // => Exit.failCause(simpleCause)

import { Cause, Channel, Effect, Exit } from "effect"

// Create a channel that fails with a lazily computed cause
let attempts = 0
const failedChannel = Channel.failCauseSync(() => {
  attempts += 1
  return Cause.fail(`Runtime error after attempt ${attempts}`)
})

const observed = [
  Effect.runSync(Effect.exit(Channel.runCollect(failedChannel))),
  attempts
] // => [Exit.fail("Runtime error after attempt 1"), 1]

import { Channel, Effect, Exit } from "effect"

let attempts = 0
const conditionalError = Channel.failSync(() => {
  attempts += 1
  return `Error after attempt ${attempts}`
})
const observed = [
  Effect.runSync(Effect.exit(Channel.runCollect(conditionalError))),
  attempts
] // => [Exit.fail("Error after attempt 1"), 1]

import { Channel, Data, Effect } from "effect"

class FlattenError extends Data.TaggedError("FlattenError")<{
  readonly cause: string
}> {}

// Create a channel that outputs channels
const nestedChannels = Channel.fromIterable([
  Channel.fromIterable([1, 2]),
  Channel.fromIterable([3, 4]),
  Channel.fromIterable([5, 6])
])

// Flatten the nested channels
const flattenedChannel = Channel.flatten(nestedChannels)

Effect.runSync(Channel.runCollect(flattenedChannel)) // => [1, 2, 3, 4, 5, 6]

import { Channel, Effect } from "effect"

const channel = Channel.fromArray([1, 2, 3, 4, 5])
Effect.runSync(Channel.runCollect(channel)) // => [1, 2, 3, 4, 5]

import { Channel, Chunk, Effect } from "effect"

const chunk = Chunk.make(1, 2, 3)
const channel = Channel.fromChunk(chunk)
Effect.runSync(Channel.runCollect(channel)) // => [1, 2, 3]

import { Channel, Effect } from "effect"

const successChannel = Channel.fromEffect(
  Effect.succeed("Hello from effect!")
)
Effect.runSync(Channel.runCollect(successChannel)) // => ["Hello from effect!"]

import { Channel, Effect } from "effect"

const set = new Set([1, 2, 3])
const channel = Channel.fromIterable(set)
Effect.runSync(Channel.runCollect(channel)) // => [1, 2, 3]

import { Channel, Effect } from "effect"

const numbers = [1, 2, 3, 4, 5]
const channel = Channel.fromIterableArray(numbers, 4)
Effect.runSync(Channel.runCollect(channel)) // => [[1, 2, 3, 4], [5]]

import { Channel, Effect } from "effect"

const numbers = [1, 2, 3, 4, 5]
const channel = Channel.fromIterator(() => numbers[Symbol.iterator]())
Effect.runSync(Channel.runCollect(channel)) // => [1, 2, 3, 4, 5]

import { Channel, Effect } from "effect"

// Create a channel from a simple iterator
const numberIterator = (): Iterator<number, string> => {
  let count = 0
  return {
    next: () => {
      if (count < 3) {
        return { value: count++, done: false }
      }
      return { value: "finished", done: true }
    }
  }
}

const channel = Channel.fromIteratorArray(() => numberIterator(), 2)
Effect.runSync(Channel.runCollect(channel)) // => [[0, 1], [2]]

import { Channel, Effect } from "effect"

// Create channel from a generator function
function* fibonacci(): Generator<number, void, unknown> {
  let a = 0, b = 1
  for (let i = 0; i < 5; i++) {
    yield a
    ;[a, b] = [b, a + b]
  }
}

const fibChannel = Channel.fromIteratorArray(() => fibonacci(), 3)
Effect.runSync(Channel.runCollect(fibChannel)) // => [[0, 1, 1], [2, 3]]

import { Channel, Data, Effect, Option, PubSub } from "effect"

class StreamError extends Data.TaggedError("StreamError")<{
  readonly message: string
}> {}

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>({ replay: 3 })

  // Create a channel that reads individual values
  const channel = Channel.fromPubSub(pubsub)

  // Publish some values
  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  yield* PubSub.publish(pubsub, 3)

  // The channel will output: 1, 2, 3 (individual values)
  return channel
})
const result = Effect.scoped(Effect.flatMap(program, Channel.runHead))
await Effect.runPromise(result) // => Option.some(1)

import { Channel, Effect, Option, PubSub } from "effect"

const notificationService = Effect.gen(function*() {
  const notificationPubSub = yield* PubSub.unbounded<string>({ replay: 1 })

  // Create a channel for real-time notifications
  const notificationChannel = Channel.fromPubSub(notificationPubSub)

  // Transform notifications to add timestamps
  const receivedAt = "2024-01-01T00:00:00.000Z"
  const timestampedChannel = Channel.map(notificationChannel, (message) => ({
    message,
    receivedAt,
    id: `notification:${message}`
  }))

  yield* PubSub.publish(notificationPubSub, "ready")
  return timestampedChannel
})
const notification = Effect.scoped(Effect.flatMap(notificationService, Channel.runHead))
await Effect.runPromise(notification) // => Option.some({ message: "ready", receivedAt: "2024-01-01T00:00:00.000Z", id: "notification:ready" })

import { Channel, Effect, Option, PubSub } from "effect"

interface DomainEvent {
  readonly type: string
  readonly payload: unknown
  readonly timestamp: number
}

const eventProcessor = Effect.gen(function*() {
  const eventPubSub = yield* PubSub.unbounded<DomainEvent>({ replay: 1 })

  // Create a channel for processing domain events
  const eventChannel = Channel.fromPubSub(eventPubSub)

  // Filter and transform events
  const processedChannel = Channel.map(eventChannel, (event) => {
    if (event.type === "user.created") {
      return {
        ...event,
        processed: true,
        processedAt: event.timestamp + 1
      }
    }
    return event
  })

  yield* PubSub.publish(eventPubSub, { type: "user.created", payload: {}, timestamp: 1 })
  return processedChannel
})
const event = Effect.scoped(Effect.flatMap(eventProcessor, Channel.runHead))
const result = await Effect.runPromise(event) // => Option.some({ type: "user.created", payload: {}, timestamp: 1, processed: true, processedAt: 2 })

import { Channel, Data, Effect, Option, PubSub } from "effect"

class BatchError extends Data.TaggedError("BatchError")<{
  readonly message: string
}> {}

const program = Effect.gen(function*() {
  const pubsub = yield* PubSub.unbounded<number>({ replay: 4 })

  // Create a channel that reads arrays of values
  const channel = Channel.fromPubSubArray(pubsub)

  // Publish some values
  yield* PubSub.publish(pubsub, 1)
  yield* PubSub.publish(pubsub, 2)
  yield* PubSub.publish(pubsub, 3)
  yield* PubSub.publish(pubsub, 4)

  // The channel will output arrays like [1, 2, 3] and [4]
  return channel
})
const result = Effect.scoped(Effect.flatMap(program, Channel.runHead))
await Effect.runPromise(result) // => Option.some([1, 2, 3, 4])

import { Channel, Effect, Option, PubSub } from "effect"

interface Order {
  readonly id: string
  readonly customerId: string
  readonly items: ReadonlyArray<string>
  readonly total: number
  readonly submittedAt: number
}

const orderBatchProcessor = Effect.gen(function*() {
  const orderPubSub = yield* PubSub.unbounded<Order>({ replay: 1 })

  // Create a channel that processes orders in batches
  const orderChannel = Channel.fromPubSubArray(orderPubSub)

  // Transform to process each batch of orders
  const processedChannel = Channel.map(orderChannel, (orderBatch) => {
    const totalRevenue = orderBatch.reduce((sum, order) => sum + order.total, 0)
    const customerCount = new Set(orderBatch.map((order) =>
      order.customerId
    )).size

    return {
      batchSize: orderBatch.length,
      totalRevenue,
      uniqueCustomers: customerCount,
      firstSubmittedAt: Math.min(...orderBatch.map((order) => order.submittedAt)),
      orders: orderBatch
    }
  })

  yield* PubSub.publish(orderPubSub, {
    id: "1", customerId: "a", items: ["book"], total: 10, submittedAt: 1
  })
  return processedChannel
})
const order = Effect.scoped(Effect.flatMap(orderBatchProcessor, Channel.runHead))
const result = await Effect.runPromise(order)
Option.map(result, (batch) => [batch.batchSize, batch.totalRevenue, batch.uniqueCustomers]) // => Option.some([1, 10, 1])

import { Channel, Effect, Option, PubSub } from "effect"

interface LogEntry {
  readonly timestamp: number
  readonly level: "info" | "warn" | "error"
  readonly message: string
  readonly source: string
}

const logAggregator = Effect.gen(function*() {
  const logPubSub = yield* PubSub.unbounded<LogEntry>({ replay: 1 })

  // Create a channel that collects logs in batches
  const logChannel = Channel.fromPubSubArray(logPubSub)

  // Transform to analyze log batches
  const analysisChannel = Channel.map(logChannel, (logBatch) => {
    const errorCount = logBatch.filter((log) => log.level === "error").length
    const warnCount = logBatch.filter((log) => log.level === "warn").length
    const infoCount = logBatch.filter((log) => log.level === "info").length

    const timeRange = {
      start: Math.min(...logBatch.map((log) => log.timestamp)),
      end: Math.max(...logBatch.map((log) => log.timestamp))
    }

    return {
      batchId: `${timeRange.start}-${timeRange.end}`,
      totalEntries: logBatch.length,
      errorCount,
      warnCount,
      infoCount,
      timeRange,
      sources: [...new Set(logBatch.map((log) => log.source))]
    }
  })

  yield* PubSub.publish(logPubSub, {
    timestamp: 1,
    level: "info",
    message: "ready",
    source: "app"
  } satisfies LogEntry)
  return analysisChannel
})
const log = Effect.scoped(Effect.flatMap(logAggregator, Channel.runHead))
const result = await Effect.runPromise(log)
Option.map(result, (batch) => [batch.batchId, batch.totalEntries, batch.infoCount]) // => Option.some(["1-1", 1, 1])

import { Cause, Channel, Effect } from "effect"

const channel = Channel.fromPull(Effect.sync(() => {
  let emitted = false
  return Effect.suspend(() => {
    if (emitted) return Cause.done()
    emitted = true
    return Effect.succeed(42)
  })
}))
await Effect.runPromise(Channel.runCollect(channel)) // => [42]

import { Cause, Channel, Effect, Queue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* Queue.bounded<string, Cause.Done>(3)
  yield* Queue.offerAll(queue, ["item1", "item2", "item3"])
  yield* Queue.end(queue)
  const channel = Channel.fromQueue(queue)
  return yield* Channel.runCollect(channel)
})
await Effect.runPromise(program) // => ["item1", "item2", "item3"]

import { Cause, Channel, Effect, Queue } from "effect"

const program = Effect.gen(function*() {
  const queue = yield* Queue.bounded<number, Cause.Done>(4)
  yield* Queue.offerAll(queue, [1, 2, 3, 4])
  yield* Queue.end(queue)
  const arrayChannel = Channel.fromQueueArray(queue)
  return yield* Channel.runCollect(arrayChannel)
})
await Effect.runPromise(program) // => [[1, 2, 3, 4]]

import { Channel, Effect } from "effect"

const channel = Channel.fromReadableStream({
  evaluate: () => new ReadableStream({
    start(controller) {
      controller.enqueue(1)
      controller.close()
    }
  }),
  onError: (cause) => new Error(String(cause))
})

await Effect.runPromise(Channel.runCollect(channel)) // => [[1]]