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
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
|
package packet
import (
"context"
"encoding/binary"
"errors"
"fmt"
"io"
"github.com/kyren223/eko/pkg/assert"
"github.com/kyren223/eko/pkg/util"
)
const framerPacketCapacity = 10
var (
PacketUnsupportedVersion error = errors.New("packet error: unsupported version")
PacketUnsupportedEncoding error = errors.New("packet error: unsupported encoding")
PacketUnsupportedType error = errors.New("packet error: unsupported type")
)
type packetFramer struct {
buffer []byte
len uint16
in chan<- Packet
inErr chan<- error
}
func RunFramer(ctx context.Context, reader io.Reader) (out <-chan Packet, outErr <-chan error) {
ch := make(chan Packet, framerPacketCapacity)
errCh := make(chan error)
framer := packetFramer{
buffer: make([]byte, PACKET_MAX_SIZE),
len: 0,
in: ch,
inErr: errCh,
}
go framer.run(ctx, util.NewChannelReader(ctx, reader))
return ch, errCh
}
func (f *packetFramer) run(ctx context.Context, reader util.ChannelReader) {
defer close(f.in)
defer close(f.inErr)
outer:
for {
select {
case data := <-reader.Out:
dataRead := 0
for dataRead < len(data) {
n := copy(f.buffer[f.len:], data[dataRead:])
assert.Assert(0 <= n && n <= int(PACKET_MAX_SIZE), "n must fit in a u16")
f.len += uint16(n)
dataRead += n
if err := f.parse(); err != nil {
f.inErr <- err
break outer
}
}
case err := <-reader.Err:
f.inErr <- err
break outer
case <-ctx.Done():
f.inErr <- ctx.Err()
break outer
}
}
}
func (f *packetFramer) parse() error {
for f.len > HEADER_SIZE {
if f.buffer[VERSION_OFFSET] != VERSION {
return fmt.Errorf("%w version=%v", PacketUnsupportedVersion, f.buffer[VERSION_OFFSET])
}
encoding := Encoding(f.buffer[ENCODING_OFFSET] >> 6)
packetType := PacketType(f.buffer[TYPE_OFFSET] & 63)
if !encoding.IsSupported() {
return PacketUnsupportedEncoding
}
if !packetType.IsSupported() {
return PacketUnsupportedType
}
length := binary.BigEndian.Uint16(f.buffer[LENGTH_OFFSET:])
if f.len-HEADER_SIZE < length {
// Wait for more data to arrive
return nil
}
fullLength := HEADER_SIZE + length
packetBuffer := make([]byte, fullLength)
copy(packetBuffer, f.buffer[:fullLength])
f.len = uint16(copy(f.buffer, f.buffer[fullLength:f.len]))
f.in <- Packet{packetBuffer}
}
return nil
}
|
