<A, E>(values: Iterable<A>): (
self: TxEnqueue<A, E>
) => Effect.Effect<Array<A>>
<A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<
Array<A>
>Offers multiple items to the queue, returning the items that were not accepted.
Details
Each item follows offer semantics: bounded queues retry while full, dropping queues reject new items when full, sliding queues evict old items to accept new items, and closing or done queues reject all items. This function mutates the original TxQueue by adding items according to the queue's strategy. It does not return a new TxQueue reference.
Example (Offering multiple values)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number>(10)
// Offer multiple items - returns rejected items as array
const rejected = yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
console.log(rejected) // [] if all accepted
console.log(rejected.length) // 0
})combinators
Source effect/TxQueue.ts:61419 lines
export const const offerAll: {
<A, E>(values: Iterable<A>): (
self: TxEnqueue<A, E>
) => Effect.Effect<Array<A>>
<A, E>(
self: TxEnqueue<A, E>,
values: Iterable<A>
): Effect.Effect<Array<A>>
}
Offers multiple items to the queue, returning the items that were not
accepted.
Details
Each item follows offer semantics: bounded queues retry while full, dropping queues reject new items when full, sliding queues evict old items to accept new items, and closing or done queues reject all items. This function mutates the original TxQueue by adding items according to the queue's strategy. It does not return a new TxQueue reference.
Example (Offering multiple values)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number>(10)
// Offer multiple items - returns rejected items as array
const rejected = yield* TxQueue.offerAll(queue, [1, 2, 3, 4, 5])
console.log(rejected) // [] if all accepted
console.log(rejected.length) // 0
})
offerAll: {
<function (type parameter) A in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>E>(values: Iterable<A>values: interface Iterable<T, TReturn = any, TNext = any>Iterable<function (type parameter) A in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>A>): (self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
interface TxEnqueue<in A, in E = never>Namespace containing type definitions for TxEnqueue variance annotations.
A TxEnqueue represents the write-only interface of a transactional queue, providing
operations for adding elements (enqueue operations) and inspecting queue state.
Example (Offering values through enqueue handles)
import { Effect, TxQueue } from "effect"
import type { Cause } from "effect"
const program = Effect.gen(function*() {
// Queue without error channel
const queue = yield* TxQueue.bounded<number>(10)
const accepted = yield* TxQueue.offer(queue, 42)
// Queue with error channel for completion signaling
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
yield* TxQueue.fail(faultTolerantQueue, "processing complete")
// Works with Done for clean completion
const completableQueue = yield* TxQueue.bounded<
string,
Cause.Done
>(5)
yield* TxQueue.offer(completableQueue, "task")
yield* TxQueue.end(completableQueue)
})
TxEnqueue<function (type parameter) A in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>E>) => import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A, E>(values: Iterable<A>): (self: TxEnqueue<A, E>) => Effect.Effect<Array<A>>A>>
<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>E>(self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
interface TxEnqueue<in A, in E = never>Namespace containing type definitions for TxEnqueue variance annotations.
A TxEnqueue represents the write-only interface of a transactional queue, providing
operations for adding elements (enqueue operations) and inspecting queue state.
Example (Offering values through enqueue handles)
import { Effect, TxQueue } from "effect"
import type { Cause } from "effect"
const program = Effect.gen(function*() {
// Queue without error channel
const queue = yield* TxQueue.bounded<number>(10)
const accepted = yield* TxQueue.offer(queue, 42)
// Queue with error channel for completion signaling
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
yield* TxQueue.fail(faultTolerantQueue, "processing complete")
// Works with Done for clean completion
const completableQueue = yield* TxQueue.bounded<
string,
Cause.Done
>(5)
yield* TxQueue.offer(completableQueue, "task")
yield* TxQueue.end(completableQueue)
})
TxEnqueue<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>E>, values: Iterable<A>values: interface Iterable<T, TReturn = any, TNext = any>Iterable<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A>>
} = import dualdual(
2,
<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>E>(self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
interface TxEnqueue<in A, in E = never>Namespace containing type definitions for TxEnqueue variance annotations.
A TxEnqueue represents the write-only interface of a transactional queue, providing
operations for adding elements (enqueue operations) and inspecting queue state.
Example (Offering values through enqueue handles)
import { Effect, TxQueue } from "effect"
import type { Cause } from "effect"
const program = Effect.gen(function*() {
// Queue without error channel
const queue = yield* TxQueue.bounded<number>(10)
const accepted = yield* TxQueue.offer(queue, 42)
// Queue with error channel for completion signaling
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offerAll(faultTolerantQueue, [1, 2, 3])
yield* TxQueue.fail(faultTolerantQueue, "processing complete")
// Works with Done for clean completion
const completableQueue = yield* TxQueue.bounded<
string,
Cause.Done
>(5)
yield* TxQueue.offer(completableQueue, "task")
yield* TxQueue.end(completableQueue)
})
TxEnqueue<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A, function (type parameter) E in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>E>, values: Iterable<A>values: interface Iterable<T, TReturn = any, TNext = any>Iterable<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<interface Array<T>Array<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A>> =>
import EffectEffect.gen(function*() {
const const rejected: A[]rejected: interface Array<T>Array<function (type parameter) A in <A, E>(self: TxEnqueue<A, E>, values: Iterable<A>): Effect.Effect<Array<A>>A> = []
for (const const value: Avalue of values: Iterable<A>values) {
const const accepted: anyaccepted = yield* const offer: {
<A, E>(value: A): (
self: TxEnqueue<A, E>
) => Effect.Effect<boolean>
<A, E>(
self: TxEnqueue<A, E>,
value: A
): Effect.Effect<boolean>
}
self: TxEnqueue<A, E>(parameter) self: {
strategy: "bounded" | "unbounded" | "dropping" | "sliding";
capacity: number;
items: TxChunk.TxChunk<any>;
stateRef: TxRef.TxRef<State<any, any>>;
toString: () => string;
toJSON: () => unknown;
}
const value: Avalue)
if (!const accepted: anyaccepted) {
const rejected: A[]rejected.Array<A>.push(...items: A[]): numberAppends new elements to the end of an array, and returns the new length of the array.
push(const value: Avalue)
}
}
return const rejected: A[]rejected
}).pipe(import EffectEffect.tx)
)