All files / packages/topic/src/reader types.ts

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
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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 }>
 
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
 
	// Cap on buffered (undelivered-to-consumer) bytes — server read credit is granted
	// against it. 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
 
	// 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 {
	// 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
}