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 | 11x 21340x 21340x 21340x 21340x 21340x 21340x 21340x | import { create, protoInt64 } from '@bufbuild/protobuf'
import {
type Duration,
DurationSchema,
type Timestamp,
timestampFromDate,
} from '@bufbuild/protobuf/wkt'
import {
type StreamReadMessage_InitRequest_TopicReadSettings,
StreamReadMessage_InitRequest_TopicReadSettingsSchema,
} from '@ydbjs/api/topic'
import type { StringValue } from 'ms'
import ms from 'ms'
import type { TopicReaderSource } from './types.js'
export let parseReadSettings = function parseReadSettings(
topic: string | TopicReaderSource | TopicReaderSource[]
): StreamReadMessage_InitRequest_TopicReadSettings[] {
let settings: StreamReadMessage_InitRequest_TopicReadSettings[] = []
let parseDuration = function parseDuration(
duration: number | StringValue | Duration
): Duration {
if (typeof duration === 'string') {
duration = ms(duration)
}
if (typeof duration === 'number') {
let seconds = Math.floor(duration / 1000)
return create(DurationSchema, {
seconds: protoInt64.parse(seconds),
nanos: (duration - seconds * 1000) * 1_000_000,
})
}
return duration
}
let parseTimestamp = function parseTimestamp(timestamp: number | Date | Timestamp): Timestamp {
if (typeof timestamp === 'number') {
timestamp = new Date(timestamp)
}
if (timestamp instanceof Date) {
timestamp = timestampFromDate(timestamp)
}
return timestamp
}
if (typeof topic === 'string') {
settings.push(
create(StreamReadMessage_InitRequest_TopicReadSettingsSchema, {
path: topic,
})
)
} else Eif (!Array.isArray(topic)) {
topic = [topic]
}
Iif (Array.isArray(topic)) {
for (let topicSource of topic) {
settings.push(
create(StreamReadMessage_InitRequest_TopicReadSettingsSchema, {
path: topicSource.path,
...(topicSource.maxLag && {
maxLag: parseDuration(topicSource.maxLag),
}),
...(topicSource.readFrom && {
readFrom: parseTimestamp(topicSource.readFrom),
}),
...(topicSource.partitionIds && {
partitionIds: topicSource.partitionIds,
}),
})
)
}
}
return settings
}
|