aboutsummaryrefslogtreecommitdiff
path: root/core/operator/harness_test.go
diff options
context:
space:
mode:
authorsangtran <tranquangsang12.7@gmail.com>2026-09-12 16:35:06 +0700
committersangtran <tranquangsang12.7@gmail.com>2026-09-12 16:35:06 +0700
commit50139c994e539f2f902fc0c06ab29ee93c1bc03c (patch)
tree7e355de76444c4eb02a178018dcc4d7ab258c466 /core/operator/harness_test.go
parenta7cb942d8b03e325b70ac95238d24939939a4c5a (diff)
signatureVerified sangtran <tranquangsang12.7@gmail.com> Key: 8DD79823D235E1BB ultimate trust
downloadflux-main.tar.gz
flux-main.zip
Verified refactor(event-time): make source timestamps authoritativeHEADmain
Diffstat (limited to 'core/operator/harness_test.go')
-rw-r--r--core/operator/harness_test.go23
1 files changed, 23 insertions, 0 deletions
diff --git a/core/operator/harness_test.go b/core/operator/harness_test.go
new file mode 100644
index 0000000..383ab8c
--- /dev/null
+++ b/core/operator/harness_test.go
@@ -0,0 +1,23 @@
1package operator_test
2
3import (
4 "context"
5
6 "flux/core"
7)
8
9type captureEmitter[T any] struct {
10 records []core.StreamRecord[T]
11}
12
13func (e *captureEmitter[T]) Emit(_ context.Context, record core.StreamRecord[T]) error {
14 e.records = append(e.records, record)
15 return nil
16}
17
18func (e *captureEmitter[T]) Records() []core.StreamRecord[T] {
19 result := make([]core.StreamRecord[T], len(e.records))
20 copy(result, e.records)
21
22 return result
23}