<A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>Adds multiple messages to the queue. Returns the remaining messages that were not added.
When to use
Use when producers can submit a batch at once and need to know which messages did not fit under the queue's capacity strategy.
Details
For bounded queues, this operation may suspend if the queue doesn't have enough capacity. The operation returns an array of messages that couldn't be added (empty array means all messages were successfully added).
Example (Offering multiple values)
import { Effect, Queue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* Queue.dropping<number>(3)
// Try to add more messages than capacity without suspending
const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
console.log(remaining1) // [4, 5] - couldn't fit the last 2
})export const const offerAll: <A, E>(
self: Enqueue<A, E>,
messages: Iterable<A>
) => Effect<Array<A>>
Adds multiple messages to the queue. Returns the remaining messages that
were not added.
When to use
Use when producers can submit a batch at once and need to know which messages
did not fit under the queue's capacity strategy.
Details
For bounded queues, this operation may suspend if the queue doesn't have
enough capacity. The operation returns an array of messages that couldn't
be added (empty array means all messages were successfully added).
Example (Offering multiple values)
import { Effect, Queue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* Queue.dropping<number>(3)
// Try to add more messages than capacity without suspending
const remaining1 = yield* Queue.offerAll(queue, [1, 2, 3, 4, 5])
console.log(remaining1) // [4, 5] - couldn't fit the last 2
})
offerAll = <function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>E>(self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
interface Enqueue<in A, in E = never>An Enqueue is a queue that can be offered to.
Details
This interface represents the write-only part of a Queue, allowing you to offer
elements to the queue but not take elements from it.
Example (Offering through enqueue handles)
import { Effect, Queue } from "effect"
// Function that only needs write access to a queue
const producer = (enqueue: Queue.Enqueue<string>) =>
Effect.gen(function*() {
yield* Queue.offer(enqueue, "hello")
yield* Queue.offerAll(enqueue, ["world", "!"])
})
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<string>(10)
yield* producer(queue)
})
Companion namespace containing type-level metadata for the Enqueue
write-only queue interface.
Enqueue<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>E>, messages: Iterable<A>messages: interface Iterable<T, TReturn = any, TNext = any>Iterable<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A>): interface Effect<out A, out E = never, out R = never>The Effect interface defines a value that lazily describes a workflow or
job. The workflow requires some context R, and may fail with an error of
type E, or succeed with a value of type A.
When to use
Use when you need to represent a lazy, composable workflow that can require
services, fail with a typed error, or succeed with a typed value.
Details
Effect values model resourceful interaction with the outside world,
including synchronous, asynchronous, concurrent, and parallel interaction.
They use a fiber-based concurrency model, with built-in support for
scheduling, fine-grained interruption, structured concurrency, and high
scalability.
To run an Effect value, you need a Runtime, which is a type that is
capable of executing Effect values.
Effect<interface Array<T>Array<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A>> =>
import internalEffectinternalEffect.const suspend: <A, E, R>(
evaluate: LazyArg<Effect.Effect<A, E, R>>
) => Effect.Effect<A, E, R>
self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
Enqueue<in A, in E = never>.state: Queue.State<any, any>state._tag: "Open" | "Closing" | "Done"_tag !== "Open") {
return import internalEffectinternalEffect.const succeed: <A>(
value: A
) => Effect.Effect<A>
import ArrArr.const fromIterable: <A>(
collection: Iterable<A>
) => A[]
Converts an Iterable to an Array.
When to use
Use to convert any Iterable (Set, Generator, etc.) into an array.
Details
If the input is already an array, this returns it by reference without
copying. Otherwise, it creates a new array from the iterable. Use copy if
you need a fresh array even when the input is already an array.
Example (Converting a Set to an array)
import { Array } from "effect"
const result = Array.fromIterable(new Set([1, 2, 3]))
console.log(result) // [1, 2, 3]
fromIterable(messages: Iterable<A>messages))
}
const const remaining: A[]remaining = const offerAllUnsafe: <A, E>(
self: Enqueue<A, E>,
messages: Iterable<A>
) => Array<A>
Adds multiple messages to the queue synchronously. Returns the remaining messages that
were not added.
When to use
Use when queue internals or a performance boundary need a synchronous batch
offer and can handle any messages that do not fit.
Gotchas
This is an unsafe operation that directly modifies the queue without Effect wrapping.
Example (Offering multiple values synchronously)
import { Cause, Effect, Queue } from "effect"
// Create a bounded queue and use unsafe API
const program = Effect.gen(function*() {
const queue = yield* Queue.bounded<number>(3)
// Try to add 5 messages to capacity-3 queue using unsafe API
const remaining = Queue.offerAllUnsafe(queue, [1, 2, 3, 4, 5])
console.log(remaining) // [4, 5] - couldn't fit the last 2
// Check what's in the queue
const size = Queue.sizeUnsafe(queue)
console.log(size) // 3
})
offerAllUnsafe(self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
interface Queue<in out A, in out E = never>A Queue is an asynchronous queue that can be offered to and taken from.
Details
It also supports signaling that it is done or failed.
Example (Offering and taking queue values)
import { Effect, Queue } from "effect"
const program = Effect.gen(function*() {
// Create a bounded queue
const queue = yield* Queue.bounded<string>(10)
// Producer: offer items to the queue
yield* Queue.offer(queue, "hello")
yield* Queue.offerAll(queue, ["world", "!"])
// Consumer: take items from the queue
const item1 = yield* Queue.take(queue)
const item2 = yield* Queue.take(queue)
const item3 = yield* Queue.take(queue)
console.log([item1, item2, item3]) // ["hello", "world", "!"]
})
Companion namespace containing type-level metadata and low-level state types
for Queue.
Queue<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>E>, messages: Iterable<A>messages)
if (const remaining: A[]remaining.Array<A>.length: numberGets or sets the length of the array. This is a number one higher than the highest index in the array.
length === 0) {
return import corecore.const exitSucceed: <A>(
a: A
) => Exit.Exit<A>
self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
Enqueue<in A, in E = never>.strategy: "suspend" | "dropping" | "sliding"strategy === "dropping") {
return import internalEffectinternalEffect.const succeed: <A>(
value: A
) => Effect.Effect<A>
const remaining: A[]remaining)
}
return const offerRemainingArray: <A, E>(
self: Enqueue<A, E>,
remaining: Array<A>
) => Effect<Array<A>, never, never>
self: Enqueue<A, E>(parameter) self: {
strategy: "suspend" | "dropping" | "sliding";
dispatcher: SchedulerDispatcher;
capacity: number;
messages: MutableList.MutableList<any>;
state: Queue.State<any, any>;
scheduleRunning: boolean;
toString: () => string;
toJSON: () => unknown;
}
interface Queue<in out A, in out E = never>A Queue is an asynchronous queue that can be offered to and taken from.
Details
It also supports signaling that it is done or failed.
Example (Offering and taking queue values)
import { Effect, Queue } from "effect"
const program = Effect.gen(function*() {
// Create a bounded queue
const queue = yield* Queue.bounded<string>(10)
// Producer: offer items to the queue
yield* Queue.offer(queue, "hello")
yield* Queue.offerAll(queue, ["world", "!"])
// Consumer: take items from the queue
const item1 = yield* Queue.take(queue)
const item2 = yield* Queue.take(queue)
const item3 = yield* Queue.take(queue)
console.log([item1, item2, item3]) // ["hello", "world", "!"]
})
Companion namespace containing type-level metadata and low-level state types
for Queue.
Queue<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Effect<Array<A>>E>, const remaining: A[]remaining)
})