Monorepo for Tangled
tangled.org
1package agentproto
2
3import (
4 "encoding/binary"
5 "fmt"
6 "io"
7 "sync"
8
9 "google.golang.org/protobuf/proto"
10
11 "buf.build/go/protovalidate"
12 agentv1 "tangled.org/core/spindle/agentproto/gen"
13)
14
15const (
16 ProtocolVersion = 1
17 DefaultPort = 10240
18 MaxMessageBytes = 1024 * 1024
19)
20
21type Message = agentv1.Message
22
23var validator protovalidate.Validator
24
25func init() {
26 var err error
27 validator, err = protovalidate.New()
28 if err != nil {
29 panic(fmt.Errorf("failed to initialize protovalidate validator: %w", err))
30 }
31}
32
33type Encoder struct {
34 mu sync.Mutex
35 w io.Writer
36}
37
38func NewEncoder(w io.Writer) *Encoder {
39 return &Encoder{w: w}
40}
41
42func (e *Encoder) Encode(msg *Message) error {
43 if err := validator.Validate(msg); err != nil {
44 return fmt.Errorf("validate agent message: %w", err)
45 }
46
47 data, err := proto.Marshal(msg)
48 if err != nil {
49 return fmt.Errorf("marshal agent message: %w", err)
50 }
51 if len(data) > MaxMessageBytes {
52 return fmt.Errorf("agent message exceeded %d bytes", MaxMessageBytes)
53 }
54
55 var header [4]byte
56 binary.BigEndian.PutUint32(header[:], uint32(len(data)))
57
58 e.mu.Lock()
59 defer e.mu.Unlock()
60 if _, err := e.w.Write(header[:]); err != nil {
61 return err
62 }
63 _, err = e.w.Write(data)
64 return err
65}
66
67type Decoder struct {
68 r io.Reader
69}
70
71func NewDecoder(r io.Reader) *Decoder {
72 return &Decoder{r: r}
73}
74
75func (d *Decoder) Decode() (*Message, error) {
76 msg := &Message{}
77 var header [4]byte
78 if _, err := io.ReadFull(d.r, header[:]); err != nil {
79 return msg, err
80 }
81
82 size := binary.BigEndian.Uint32(header[:])
83 if size > MaxMessageBytes {
84 return msg, fmt.Errorf("agent message exceeded %d bytes", MaxMessageBytes)
85 }
86
87 data := make([]byte, size)
88 if _, err := io.ReadFull(d.r, data); err != nil {
89 return msg, err
90 }
91 if err := proto.Unmarshal(data, msg); err != nil {
92 return msg, fmt.Errorf("parse agent message: %w", err)
93 }
94 if err := validator.Validate(msg); err != nil {
95 return msg, fmt.Errorf("validate agent message: %w", err)
96 }
97 return msg, nil
98}