<A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>Waits transactionally for the next item and returns it without removing it.
Details
If the queue is open but empty, the transaction retries until an item is available or the queue completes. If the queue is done, the queue's completion cause is propagated through the error channel.
Example (Peeking without removing values)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offer(queue, 42)
// Peek at the next item without removing it
const item = yield* TxQueue.peek(queue)
console.log(item) // 42
// Item is still in the queue
const size = yield* TxQueue.size(queue)
console.log(size) // 1
})
// Error handling example
const errorExample = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number, string>(5)
yield* TxQueue.fail(queue, "queue failed")
// peek() propagates the queue error through E-channel
const result = yield* Effect.flip(TxQueue.peek(queue))
console.log(result) // "queue failed"
})combinators
Source effect/TxQueue.ts:101915 lines
export const const peek: <A, E>(
self: TxDequeue<A, E>
) => Effect.Effect<A, E>
Waits transactionally for the next item and returns it without removing it.
Details
If the queue is open but empty, the transaction retries until an item is available or the queue completes. If the queue is done, the queue's completion cause is propagated through the error channel.
Example (Peeking without removing values)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.offer(queue, 42)
// Peek at the next item without removing it
const item = yield* TxQueue.peek(queue)
console.log(item) // 42
// Item is still in the queue
const size = yield* TxQueue.size(queue)
console.log(size) // 1
})
// Error handling example
const errorExample = Effect.gen(function*() {
const queue = yield* TxQueue.bounded<number, string>(5)
yield* TxQueue.fail(queue, "queue failed")
// peek() propagates the queue error through E-channel
const result = yield* Effect.flip(TxQueue.peek(queue))
console.log(result) // "queue failed"
})
peek = <function (type parameter) A in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>A, function (type parameter) E in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>E>(self: TxDequeue<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 TxDequeue<out A, out E = never>Namespace containing type definitions for TxDequeue variance annotations.
A TxDequeue represents the read-only interface of a transactional queue, providing
operations for consuming elements (dequeue operations) and inspecting queue state.
Example (Taking values through dequeue handles)
import { Effect, TxQueue } from "effect"
const program = Effect.gen(function*() {
// Queue without error channel
const queue = yield* TxQueue.bounded<number>(10)
yield* TxQueue.offer(queue, 42)
const item = yield* TxQueue.take(queue)
console.log(item) // 42
// Queue with error channel - errors propagate through E-channel
const faultTolerantQueue = yield* TxQueue.bounded<number, string>(10)
yield* TxQueue.fail(faultTolerantQueue, "processing failed")
// All dequeue operations now fail with the error directly
const takeResult = yield* Effect.flip(TxQueue.take(faultTolerantQueue)) // "processing failed"
const peekResult = yield* Effect.flip(TxQueue.peek(faultTolerantQueue)) // "processing failed"
})
TxDequeue<function (type parameter) A in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>A, function (type parameter) E in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>E>): import EffectEffect.type Effect.Effect = /*unresolved*/ anyEffect<function (type parameter) A in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>A, function (type parameter) E in <A, E>(self: TxDequeue<A, E>): Effect.Effect<A, E>E> =>
import EffectEffect.gen(function*() {
const const state: State<any, any>state = yield* import TxRefTxRef.const get: <A>(
self: TxRef<A>
) => Effect.Effect<A>
Reads the current value of the TxRef.
When to use
Use to read the current value of a TxRef.
Example (Reading transactional references)
import { Effect, TxRef } from "effect"
const program = Effect.gen(function*() {
const counter = yield* TxRef.make(42)
// Read the value within a transaction
const value = yield* Effect.tx(
TxRef.get(counter)
)
console.log(value) // 42
})
get(self: TxDequeue<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;
}
TxQueueState.stateRef: TxRef.TxRef<State<any, any>>(property) TxQueueState.stateRef: {
version: number;
pending: Map<unknown, () => void>;
value: 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; <…;
}
const state: State<any, any>state._tag === "Done") {
return yield* import EffectEffect.failCause(const state: anyconst state: {
_tag: "Done";
cause: Cause.Cause<E>;
}
const chunk: Chunk.Chunk<any>const chunk: {
length: number;
right: Chunk<A>;
left: Chunk<A>;
backing: Backing<A>;
depth: number;
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; <…;
toString: () => string;
toJSON: () => unknown;
}
import TxChunkTxChunk.get(self: TxDequeue<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;
}
TxQueueState.items: TxChunk.TxChunk<any>(property) TxQueueState.items: {
ref: TxRef.TxRef<Chunk.Chunk<A>>;
toString: () => string;
toJSON: () => unknown;
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; <…;
}
const head: Option.Option<any>head = import ChunkChunk.head(const chunk: Chunk.Chunk<any>const chunk: {
length: number;
right: Chunk<A>;
left: Chunk<A>;
backing: Backing<A>;
depth: number;
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; <…;
toString: () => string;
toJSON: () => unknown;
}
import OptionOption.isNone(const head: Option.Option<any>head)) {
return yield* import EffectEffect.txRetry
}
return const head: Option.Some<any>const head: {
_tag: "Some";
_op: "Some";
value: A;
valueOrUndefined: 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; <…;
toString: () => string;
toJSON: () => unknown;
}
import EffectEffect.tx)