<A>(pubsub: PubSub.PubSub<A>): Stream<A>Creates a stream from a subscription to a PubSub.
Example (Creating a stream from a subscription to a PubSub)
import { Console, Effect, Fiber, PubSub, Stream } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.unbounded<number>()
const fiber = yield* Stream.fromPubSub(pubsub).pipe(
Stream.take(3),
Stream.runCollect,
Effect.forkChild
)
yield* PubSub.publish(pubsub, 1)
yield* PubSub.publish(pubsub, 2)
yield* PubSub.publish(pubsub, 3)
const values = yield* Fiber.join(fiber)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]constructors
Source effect/Stream.ts:13281 lines
export const const fromPubSub: <A>(
pubsub: PubSub.PubSub<A>
) => Stream<A>
Creates a stream from a subscription to a PubSub.
Example (Creating a stream from a subscription to a PubSub)
import { Console, Effect, Fiber, PubSub, Stream } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.unbounded<number>()
const fiber = yield* Stream.fromPubSub(pubsub).pipe(
Stream.take(3),
Stream.runCollect,
Effect.forkChild
)
yield* PubSub.publish(pubsub, 1)
yield* PubSub.publish(pubsub, 2)
yield* PubSub.publish(pubsub, 3)
const values = yield* Fiber.join(fiber)
yield* Console.log(values)
})
Effect.runPromise(program)
// Output: [ 1, 2, 3 ]
fromPubSub = <function (type parameter) A in <A>(pubsub: PubSub.PubSub<A>): Stream<A>A>(pubsub: PubSub.PubSub<A>(parameter) pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
import PubSubPubSub.interface PubSub<in out A>A PubSub<A> is an asynchronous message hub into which publishers can publish
messages of type A and subscribers can subscribe to take messages of type
A.
Example (Publishing and subscribing to messages)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
// Create a bounded PubSub with capacity 10
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe and consume messages
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Publish messages
yield* PubSub.publish(pubsub, "Hello")
yield* PubSub.publish(pubsub, "World")
const message1 = yield* PubSub.take(subscription)
const message2 = yield* PubSub.take(subscription)
console.log(message1, message2) // "Hello", "World"
}))
})
Companion namespace containing the low-level building blocks used by
PubSub, including atomic implementations, backing subscriptions, replay
windows, and delivery strategies.
PubSub<function (type parameter) A in <A>(pubsub: PubSub.PubSub<A>): Stream<A>A>): interface Stream<out A, out E = never, out R = never>A Stream<A, E, R> describes a program that can emit many A values, fail
with E, and require R.
Details
Streams are pull-based with backpressure and emit chunks to amortize effect
evaluation. They support monadic composition and error handling similar to
Effect, adapted for multiple values.
Example (Creating and consuming streams)
import { Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
yield* Stream.make(1, 2, 3).pipe(
Stream.map((n) => n * 2),
Stream.runForEach((n) => Console.log(n))
)
})
Effect.runPromise(program)
// Output:
// 2
// 4
// 6
Stream<function (type parameter) A in <A>(pubsub: PubSub.PubSub<A>): Stream<A>A> => const fromChannel: <
Arr extends Arr.NonEmptyReadonlyArray<any>,
E,
R
>(
channel: Channel.Channel<
Arr,
E,
void,
unknown,
unknown,
unknown,
R
>
) => Stream<
Arr extends Arr.NonEmptyReadonlyArray<infer A>
? A
: never,
E,
R
>
Creates a stream from a array-emitting Channel.
Example (Creating a stream from an array-emitting channel)
import { Channel, Console, Effect, Stream } from "effect"
const program = Effect.gen(function*() {
const channel = Channel.succeed([1, 2, 3] as const)
const stream = Stream.fromChannel(channel)
const result = yield* Stream.runCollect(stream)
yield* Console.log(result)
})
// Output: [ 1, 2, 3 ]
fromChannel(import ChannelChannel.const fromPubSubArray: <A>(
pubsub: PubSub.PubSub<A>
) => Channel<Arr.NonEmptyReadonlyArray<A>>
Creates a channel from a PubSub that outputs arrays of values.
Details
This constructor creates a channel that reads from a PubSub by automatically
subscribing to it and collecting values into arrays. The channel outputs
arrays of values in chunks, making it ideal for batch processing scenarios.
Example (Batching PubSub values)
import { Channel, Data, Effect, PubSub } from "effect"
class BatchError extends Data.TaggedError("BatchError")<{
readonly message: string
}> {}
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<number>(16)
// 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
})
Example (Processing PubSub orders in batches)
import { Channel, Effect, 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.bounded<Order>(100)
// 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
}
})
return processedChannel
})
Example (Processing PubSub logs in batches)
import { Channel, Effect, 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.bounded<LogEntry>(500)
// 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))]
}
})
return analysisChannel
})
fromPubSubArray(pubsub: PubSub.PubSub<A>(parameter) pubsub: {
pubsub: PubSub.Atomic<A>;
subscribers: PubSub.Subscribers<A>;
scope: Scope.Closeable;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<A>;
pipe: { <A>(this: A): A; <A, B = never>(this: A, ab: (_: A) => B): B; <A, B = never, C = never>(this: A, ab: (_: A) => B, bc: (_: B) => C): C; <A, B = never, C = never, D = never>(this: A, ab: (_: A) => B, bc: (_: B) => C, cd: (_: C) => D): D; <…;
}
Referenced by 1 symbols