<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
})Offering
Source effect/Queue.ts:79533 lines
export const 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 = <function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): 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>): Array<A>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): 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>): Array<A>A>): interface Array<T>Array<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A>A> => {
if (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 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)
} else if (
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>.capacity: numbercapacity === var Number: NumberConstructorAn object that represents a number of any kind. All JavaScript numbers are 64-bit floating-point numbers.
Number.NumberConstructor.POSITIVE_INFINITY: numberA value greater than the largest number that can be represented in JavaScript.
JavaScript displays POSITIVE_INFINITY values as infinity.
POSITIVE_INFINITY ||
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 === "sliding"
) {
import MutableListMutableList.const appendAll: <A>(
self: MutableList<A>,
messages: Iterable<A>
) => number
Appends all elements from an iterable to the end of the MutableList.
Returns the number of elements added.
Example (Appending multiple elements)
import { MutableList } from "effect"
const list = MutableList.make<number>()
MutableList.append(list, 1)
MutableList.append(list, 2)
// Append multiple elements
const added = MutableList.appendAll(list, [3, 4, 5])
console.log(added) // 3
console.log(list.length) // 5
// Elements maintain order: [1, 2, 3, 4, 5]
console.log(MutableList.takeAll(list)) // [1, 2, 3, 4, 5]
// Works with any iterable
const newList = MutableList.make<string>()
MutableList.appendAll(newList, new Set(["a", "b", "c"]))
console.log(MutableList.takeAll(newList)) // ["a", "b", "c"]
// Useful for bulk loading
const bulkList = MutableList.make<number>()
const count = MutableList.appendAll(
bulkList,
Array.from({ length: 1000 }, (_, i) => i)
)
console.log(count) // 1000
appendAll(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>.messages: MutableList.MutableList<any>(property) Enqueue<in A, in E = never>.messages: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
messages: Iterable<A>messages)
if (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 === "sliding") {
import MutableListMutableList.const takeN: <A>(
self: MutableList<A>,
n: number
) => Array<A>
Takes up to N elements from the beginning of the MutableList and returns them as an array.
The taken elements are removed from the list. This operation is optimized for performance
and includes zero-copy optimizations when possible.
Example (Taking batches)
import { MutableList } from "effect"
const list = MutableList.make<number>()
MutableList.appendAll(list, [1, 2, 3, 4, 5, 6, 7, 8, 9, 10])
console.log(list.length) // 10
// Take first 3 elements
const first3 = MutableList.takeN(list, 3)
console.log(first3) // [1, 2, 3]
console.log(list.length) // 7
// Take more than available
const remaining = MutableList.takeN(list, 20)
console.log(remaining) // [4, 5, 6, 7, 8, 9, 10]
console.log(list.length) // 0
// Take from empty list
const empty = MutableList.takeN(list, 5)
console.log(empty) // []
// Batch processing pattern
const queue = MutableList.make<string>()
MutableList.appendAll(queue, ["task1", "task2", "task3", "task4", "task5"])
while (queue.length > 0) {
const batch = MutableList.takeN(queue, 2) // Process 2 at a time
console.log("Processing batch:", batch)
}
takeN(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>.messages: MutableList.MutableList<any>(property) Enqueue<in A, in E = never>.messages: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
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>.messages: MutableList.MutableList<any>(property) Enqueue<in A, in E = never>.messages: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
MutableList<any>.length: numberlength - 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>.capacity: numbercapacity)
}
const scheduleReleaseTaker: <A, E>(
self: Enqueue<A, E>
) => void
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>): Array<A>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A>E>)
return []
}
const const free: numberfree = 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>.capacity: numbercapacity <= 0
? 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: {
readonly _tag: "Open";
readonly takers: Set<(_: Effect<void, any, never>) => void>;
readonly offers: Set<Queue.OfferEntry<any>>;
readonly awaiters: Set<(_: Effect<void, any, never>) => void>;
}
takers: Set<(_: Effect<void, E>) => void>takers.Set<(_: Effect<void, any, never>) => void>.size: numbersize
: 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>.capacity: numbercapacity - 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>.messages: MutableList.MutableList<any>(property) Enqueue<in A, in E = never>.messages: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
MutableList<any>.length: numberlength
if (const free: numberfree === 0) {
return 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: interface Array<T>Array<function (type parameter) A in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A>A> = []
let let i: numberi = 0
for (const const message: Amessage of messages: Iterable<A>messages) {
if (let i: numberi < const free: numberfree) {
import MutableListMutableList.const append: <A>(
self: MutableList<A>,
message: A
) => void
Appends an element to the end of the MutableList.
This operation is optimized for high-frequency usage.
Example (Appending elements)
import { MutableList } from "effect"
const list = MutableList.make<number>()
// Append elements one by one
MutableList.append(list, 1)
MutableList.append(list, 2)
MutableList.append(list, 3)
console.log(list.length) // 3
// Elements are taken from head (FIFO)
console.log(MutableList.take(list)) // 1
console.log(MutableList.take(list)) // 2
console.log(MutableList.take(list)) // 3
// High-throughput usage
for (let i = 0; i < 10000; i++) {
MutableList.append(list, i)
}
append(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>.messages: MutableList.MutableList<any>(property) Enqueue<in A, in E = never>.messages: {
head: MutableList.Bucket<A> | undefined;
tail: MutableList.Bucket<A> | undefined;
length: number;
}
const message: Amessage)
} else {
const remaining: A[]remaining.Array<A>.push(...items: A[]): numberAppends new elements to the end of an array, and returns the new length of the array.
push(const message: Amessage)
}
let i: numberi++
}
const scheduleReleaseTaker: <A, E>(
self: Enqueue<A, E>
) => void
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>): Array<A>A, function (type parameter) E in <A, E>(self: Enqueue<A, E>, messages: Iterable<A>): Array<A>E>)
return const remaining: A[]remaining
}Referenced by 2 symbols