blob: cb160b2386c2e1fd4b78c784c77ef7abb2529e61 (
plain)
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
|
package core
type RecordType byte
const (
RecordData RecordType = iota
RecordWatermark
RecordBarrier
)
type StreamRecord[T any] struct {
Value T
Key string // for keyBy stateful processing
Timestamp int64 // epoch ms time
BarrierID uint64 // ID checkpoint barrier
Type RecordType
}
func NewDataRecord[T any](key string, val T, timestamp int64) *StreamRecord[T] {
return &StreamRecord[T]{
Value: val,
Key: key,
Timestamp: timestamp,
Type: RecordData,
}
}
func NewWatermarkRecord[T any](timestamp int64) *StreamRecord[T] {
return &StreamRecord[T]{
Type: RecordWatermark,
Timestamp: timestamp,
}
}
func NewBarrierRecord[T any](barrierID uint64) *StreamRecord[T] {
return &StreamRecord[T]{
Type: RecordBarrier,
BarrierID: barrierID,
}
}
|