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 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 | 24x 3468x 3468x 1646x 3468x 3468x 24x 86728x 86728x 4x 86724x 11537x 11535x 2x 1x 1x 49776x 257x 49519x 14663x 14663x 34856x 20378x 20378x 14478x 1822x 12656x 1646x 11010x 11010x 3469x 3244x 225x 224x 1x 21942x | import { fail } from 'node:assert/strict'
import type { StreamReadMessage_FromServer } from '@ydbjs/api/topic'
import type { TransitionResult, TransitionRuntime } from '@ydbjs/fsm'
// One physical streamRead lifecycle. All reader business logic (partition
// sessions, offsets, commits, buffer, backoff) lives in the reader FSM; the
// transport only owns the stream and forwards server messages as outputs.
//
// The full transition map (table + diagram) lives in packages/topic/ARCHITECTURE.md —
// update it in the same commit when you change this dispatch.
//
// Unlike the writer transport, the reader has many server-message types
// (read / start-partition / stop-partition / commit / end-partition / …), so
// the transport does not classify them — it recognizes only the init handshake
// and forwards every other server message verbatim for the reader FSM to route.
export type TransportState = 'idle' | 'connecting' | 'ready' | 'disconnected' | 'closed'
// The transport FSM keeps no ctx state — every fact rides on the event/output.
export type TransportCtx = {}
export type TransportEvent =
// lifecycle commands the owner dispatches inward
| { type: 'transport.connect' }
| { type: 'transport.close' }
| { type: 'transport.destroy'; reason?: unknown }
// stream facts the ingest task forwards — only the init handshake is
// recognized; every other server message rides verbatim in transport.message
// for the reader FSM to route
| { type: 'transport.init'; sessionId: string }
| { type: 'transport.message'; message: StreamReadMessage_FromServer }
| { type: 'transport.ended' }
| { type: 'transport.error'; error: unknown }
export type TransportEffect =
| { type: 'transport.effect.open_stream' }
| { type: 'transport.effect.close_stream' }
| { type: 'transport.effect.finalize'; reason: unknown }
// Emitted to the reader FSM, which ingests these as reader events.
export type TransportOutput =
| { type: 'transport.stream.init_response'; sessionId: string }
| { type: 'transport.stream.message'; message: StreamReadMessage_FromServer }
| { type: 'transport.stream.disconnected'; error?: unknown }
type TransportRuntime = TransitionRuntime<TransportState, TransportEvent, TransportOutput>
let disconnect = function disconnect(
error: unknown,
runtime: TransportRuntime
): TransitionResult<TransportState, TransportEffect> {
let output: TransportOutput = { type: 'transport.stream.disconnected' }
if (error !== undefined) {
output = { ...output, error }
}
runtime.emit(output)
return { state: 'disconnected', effects: [{ type: 'transport.effect.close_stream' }] }
}
export let transportTransition = function transportTransition(
_ctx: TransportCtx,
event: TransportEvent,
runtime: TransportRuntime
): TransitionResult<TransportState, TransportEffect> | void {
let state = runtime.state
if (state !== 'closed' && event.type === 'transport.destroy') {
return {
state: 'closed',
effects: [
{ type: 'transport.effect.close_stream' },
{
type: 'transport.effect.finalize',
reason: event.reason ?? new Error('Transport destroyed'),
},
],
}
}
switch (state) {
case 'idle': {
if (event.type === 'transport.connect') {
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'connecting':
case 'ready': {
if (event.type === 'transport.connect') {
// Reopen the stream (open_stream disposes the previous one first).
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.init') {
runtime.emit({ type: 'transport.stream.init_response', sessionId: event.sessionId })
return state === 'ready' ? undefined : { state: 'ready' }
}
if (event.type === 'transport.message') {
runtime.emit({ type: 'transport.stream.message', message: event.message })
return
}
if (event.type === 'transport.ended') {
return disconnect(undefined, runtime)
}
if (event.type === 'transport.error') {
return disconnect(event.error, runtime)
}
Eif (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{ type: 'transport.effect.close_stream' },
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'disconnected': {
if (event.type === 'transport.connect') {
return { state: 'connecting', effects: [{ type: 'transport.effect.open_stream' }] }
}
if (event.type === 'transport.close') {
return {
state: 'closed',
effects: [
{
type: 'transport.effect.finalize',
reason: new Error('Transport closed'),
},
],
}
}
return
}
case 'closed':
return
default: {
let exhaustive: never = state
fail(`Unhandled transport state: ${exhaustive}`)
}
}
}
|