blob: 4ac2c834c87eed5231564d0ab4344e551fa44569 (
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
|
package transport
import "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() (core.StreamRecord[T], bool) {
val, ok := <-c.recordCh
return val, ok
}
func (c *LocalReceiveChannel[T]) Close() error {
// The emitter will close
return nil
}
type LocalEmitChannel[T any] struct {
recordCh chan<- core.StreamRecord[T]
}
func (l LocalEmitChannel[T]) Emit(c core.StreamRecord[T]) error {
l.recordCh <- c
return nil
}
func (l LocalEmitChannel[T]) Close() error {
close(l.recordCh)
return nil
}
|