Press n or j to go to the next uncovered block, b, p or k for the previous block.
| 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 | import type { Duration, Timestamp } from '@bufbuild/protobuf/wkt'
import type { StringValue } from 'ms'
import type { CodecMap } from '../codec.js'
import type { TopicMessage } from '../message.js'
import type { TopicPartitionSession } from '../partition-session.js'
export type TopicReaderSource = {
// Topic path.
path: string
// Partitions to read from this topic; an empty / omitted list reads them all.
partitionIds?: bigint[]
// Skip messages older than now - maxLag. A bare number is milliseconds; zero
// means infinite lag.
maxLag?: number | StringValue | Duration
// Read only messages written at or after this time. A bare number is epoch ms.
readFrom?: number | Date | Timestamp
}
export type onPartitionSessionStartCallback = (
partitionSession: TopicPartitionSession,
committedOffset: bigint,
partitionOffsets: {
start: bigint
end: bigint
}
) => Promise<void | undefined | { readOffset?: bigint; commitOffset?: bigint }>
// On a graceful (soft) stop the callback runs while the session is still
// deliverable and committable, and is awaited before the stop response is sent —
// the last chance to commit processed offsets before the partition moves to
// another reader (bounded by gracefulShutdownTimeoutMs). On a forced stop or
// end-of-partition it is informational: the session is already stopped.
export type onPartitionSessionStopCallback = (
partitionSession: TopicPartitionSession,
committedOffset: bigint
) => Promise<void>
export type onCommittedOffsetCallback = (
partitionSession: TopicPartitionSession,
committedOffset: bigint
) => void
export type TopicReaderOptions = {
// Topic path, or one/many sources with per-topic partitionIds/maxLag/readFrom filters.
topic: string | TopicReaderSource | TopicReaderSource[]
// Consumer the read session is attributed to — the server tracks committed
// offsets per consumer.
consumer: string
// Codecs available for decompression. Defaults to RAW/GZIP/ZSTD (defaultCodecMap).
codecMap?: CodecMap
// Initial server read-credit window. The server may exceed it for an oversized
// message; consumed response bytes are granted back exactly. Default 8MiB.
maxBufferBytes?: bigint
// How often to refresh the auth token on the stream. Default 60s.
updateTokenIntervalMs?: number
// Force-close deadline for graceful close() before pending commits are dropped.
// JS-specific safety net; other SDKs bound this by the caller's signal only. Default 30s.
gracefulShutdownTimeoutMs?: number
// Terminal reconnect window in ms. Unbounded by default (the reader reconnects
// forever, waiting for the server / topic); pass a finite value to fail terminally
// if no successful reconnect happens within it.
recoveryWindowMs?: number
// Retry on SCHEME_ERROR (e.g. the topic does not exist yet). Off by default: a
// missing / mistyped topic fails fast. Enable to wait until the topic is created.
// A running reader whose topic is dropped idles until the server closes the stale
// stream (~1 min), then transparently reconnects: it resumes automatically if the
// topic exists again, and with this flag also waits if the topic is recreated later.
retryOnSchemeError?: boolean
// Declare autopartitioning support to the server: on topics with partition
// autoscaling the server then ends fully-read partitions via EndPartitionSession
// (child ids appear on TopicPartitionSession.childPartitionIds) instead of
// heuristic rebalancing. Off by default.
autoPartitioningSupport?: boolean
// Called when the server assigns a partition; its return value may override
// read/commit offsets.
onPartitionSessionStart?: onPartitionSessionStartCallback
// Called when the server revokes a partition — last chance to commit its offsets.
onPartitionSessionStop?: onPartitionSessionStopCallback
// Called after the server acknowledges a commit.
onCommittedOffset?: onCommittedOffsetCallback
}
// Options for one read() loop — shared by TopicReader.read() and TopicTxReader.read().
export type TopicReadOptions = {
limit?: number
// Max time to accumulate a batch before yielding (possibly empty, so an idle
// topic never hangs the consumer). Omit to block for one chunk.
batchWindowMs?: number
/** @deprecated renamed to `batchWindowMs`. */
waitMs?: number
signal?: AbortSignal
}
// The public reader type is the `TopicReader` class itself (see reader/reader.ts),
// mirroring `TopicWriter` — there is no separate interface. `TopicTxReader` stays an
// interface because a tx reader is a distinct shape (no `commit()`).
export interface TopicTxReader extends AsyncDisposable, Disposable {
// Server-accounted bytes retained until their responses pass through read().
readonly bufferedBytes: bigint
// Read messages from the topic stream within a transaction.
read(options?: TopicReadOptions): AsyncIterable<TopicMessage[]>
// Gracefully close the reader.
close(): Promise<void>
// Immediately destroy the reader and release all resources.
destroy(reason?: unknown): void
}
|