blob: e3e4cba76cda63355e55bcaa28df9ee89e8c685c (
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
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
|
package transport
import (
"context"
"goflink/core"
)
type LocalReceiveChannel[T any] struct {
recordCh <-chan core.StreamRecord[T]
}
func NewLocalReceiveChannel[T any](ch <-chan core.StreamRecord[T]) *LocalReceiveChannel[T] {
return &LocalReceiveChannel[T]{
recordCh: ch,
}
}
func (c *LocalReceiveChannel[T]) Receive(ctx context.Context) (core.StreamRecord[T], bool) {
select {
case val, ok := <-c.recordCh:
return val, ok
case <-ctx.Done():
var zero core.StreamRecord[T]
return zero, false
}
}
func (c *LocalReceiveChannel[T]) Close() error {
// The emitter will close
return nil
}
type LocalEmitChannel[T any] struct {
recordCh chan<- core.StreamRecord[T]
}
func NewLocalEmitChannel[T any](ch chan<- core.StreamRecord[T]) *LocalEmitChannel[T] {
return &LocalEmitChannel[T]{
recordCh: ch,
}
}
func (l *LocalEmitChannel[T]) Emit(ctx context.Context, c core.StreamRecord[T]) error {
select {
case l.recordCh <- c:
return nil
case <-ctx.Done():
return ctx.Err()
}
}
func (l *LocalEmitChannel[T]) Close() error {
close(l.recordCh)
return nil
}
|