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 | 13x 700x 700x 347x 700x 700x 13x 60426x 60426x 4x 60422x 11473x 11471x 2x 1x 1x 26128x 72x 26056x 11660x 11660x 14396x 2495x 2495x 11901x 353x 11548x 347x 11201x 11201x 701x 428x 273x 272x 1x 22120x | 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}`)
}
}
}
|