(min: number, max: number): <A>(
self: Subscription<A>
) => Effect.Effect<Array<A>>
<A>(self: Subscription<A>, min: number, max: number): Effect.Effect<
Array<A>
>Takes between the specified minimum and maximum number of messages from the subscription. Will suspend if the minimum number is not immediately available.
Example (Taking between a minimum and maximum)
import { Effect, Fiber, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Start taking between 2 and 5 messages (will suspend)
const takeFiber = yield* Effect.forkChild(
PubSub.takeBetween(subscription, 2, 5)
)
// Publish 3 messages
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Now the take will complete with 3 messages
const messages = yield* Fiber.join(takeFiber)
console.log("Between 2-5:", messages) // ["msg1", "msg2", "msg3"]
}))
})subscriptions
Source effect/PubSub.ts:13408 lines
export const const takeBetween: {
(min: number, max: number): <A>(
self: Subscription<A>
) => Effect.Effect<Array<A>>
<A>(
self: Subscription<A>,
min: number,
max: number
): Effect.Effect<Array<A>>
}
Takes between the specified minimum and maximum number of messages from the subscription.
Will suspend if the minimum number is not immediately available.
Example (Taking between a minimum and maximum)
import { Effect, Fiber, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
yield* Effect.scoped(Effect.gen(function*() {
const subscription = yield* PubSub.subscribe(pubsub)
// Start taking between 2 and 5 messages (will suspend)
const takeFiber = yield* Effect.forkChild(
PubSub.takeBetween(subscription, 2, 5)
)
// Publish 3 messages
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Now the take will complete with 3 messages
const messages = yield* Fiber.join(takeFiber)
console.log("Between 2-5:", messages) // ["msg1", "msg2", "msg3"]
}))
})
takeBetween: {
(min: numbermin: number, max: numbermax: number): <function (type parameter) A in <A>(self: Subscription<A>): Effect.Effect<Array<A>>A>(self: Subscription<A>(parameter) self: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<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; <…;
}
interface Subscription<out A>A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Example (Taking messages from a subscription)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe within a scope for automatic cleanup
yield* Effect.scoped(Effect.gen(function*() {
const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(
pubsub
)
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Take individual messages
const message = yield* PubSub.take(subscription)
console.log(message) // "msg1"
// Take multiple messages
const messages = yield* PubSub.takeUpTo(subscription, 1)
console.log(messages) // ["msg2"]
const allMessages = yield* PubSub.takeAll(subscription)
console.log(allMessages) // ["msg3"]
}))
})
Subscription<function (type parameter) A in <A>(self: Subscription<A>): Effect.Effect<Array<A>>A>) => import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A>(self: Subscription<A>): Effect.Effect<Array<A>>A>>
<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>(self: Subscription<A>(parameter) self: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<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; <…;
}
interface Subscription<out A>A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Example (Taking messages from a subscription)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe within a scope for automatic cleanup
yield* Effect.scoped(Effect.gen(function*() {
const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(
pubsub
)
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Take individual messages
const message = yield* PubSub.take(subscription)
console.log(message) // "msg1"
// Take multiple messages
const messages = yield* PubSub.takeUpTo(subscription, 1)
console.log(messages) // ["msg2"]
const allMessages = yield* PubSub.takeAll(subscription)
console.log(allMessages) // ["msg3"]
}))
})
Subscription<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>, min: numbermin: number, max: numbermax: number): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>>
} = import dualdual(
3,
<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>(self: Subscription<A>(parameter) self: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<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; <…;
}
interface Subscription<out A>A subscription represents a consumer's connection to a PubSub, allowing them to take messages.
Example (Taking messages from a subscription)
import { Effect, PubSub } from "effect"
const program = Effect.gen(function*() {
const pubsub = yield* PubSub.bounded<string>(10)
// Subscribe within a scope for automatic cleanup
yield* Effect.scoped(Effect.gen(function*() {
const subscription: PubSub.Subscription<string> = yield* PubSub.subscribe(
pubsub
)
yield* PubSub.publishAll(pubsub, ["msg1", "msg2", "msg3"])
// Take individual messages
const message = yield* PubSub.take(subscription)
console.log(message) // "msg1"
// Take multiple messages
const messages = yield* PubSub.takeUpTo(subscription, 1)
console.log(messages) // ["msg2"]
const allMessages = yield* PubSub.takeAll(subscription)
console.log(allMessages) // ["msg3"]
}))
})
Subscription<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>, min: numbermin: number, max: numbermax: number): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A>(self: Subscription<A>, min: number, max: number): Effect.Effect<Array<A>>A>> =>
import EffectEffect.suspend(() => const takeRemainderLoop: <A>(
self: Subscription<A>,
min: number,
max: number,
acc: Array<A>
) => Effect.Effect<Array<A>>
self: Subscription<A>(parameter) self: {
pubsub: PubSub.Atomic<any>;
subscribers: PubSub.Subscribers<any>;
subscription: PubSub.BackingSubscription<A>;
pollers: MutableList.MutableList<Deferred.Deferred<any>>;
shutdownHook: Latch.Latch;
shutdownFlag: MutableRef.MutableRef<boolean>;
strategy: PubSub.Strategy<any>;
replayWindow: PubSub.ReplayWindow<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; <…;
}
min: numbermin, max: numbermax, []))
)