~james/core

core/spindle/agentproto/protocol.go -rw-r--r-- 2.1 KiB
452f17b7 — james chore(import): snapshot james.dad.town/core at f70c71c7696a 15 hours ago
                                                                                
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
package agentproto

import (
	"encoding/binary"
	"fmt"
	"io"
	"sync"

	"google.golang.org/protobuf/proto"

	"buf.build/go/protovalidate"
	agentv1 "tangled.org/core/spindle/agentproto/gen"
)

const (
	ProtocolVersion = 2
	DefaultPort     = 10240
	MaxMessageBytes = 1024 * 1024
)

type Message = agentv1.Message

func ValidateProtocolVersion(got uint32) error {
	if got != ProtocolVersion {
		return fmt.Errorf("agent protocol version mismatch: got %d, want %d", got, ProtocolVersion)
	}
	return nil
}

var validator protovalidate.Validator

func init() {
	var err error
	validator, err = protovalidate.New()
	if err != nil {
		panic(fmt.Errorf("failed to initialize protovalidate validator: %w", err))
	}
}

type Encoder struct {
	mu sync.Mutex
	w  io.Writer
}

func NewEncoder(w io.Writer) *Encoder {
	return &Encoder{w: w}
}

func (e *Encoder) Encode(msg *Message) error {
	if err := validator.Validate(msg); err != nil {
		return fmt.Errorf("validate agent message: %w", err)
	}

	data, err := proto.Marshal(msg)
	if err != nil {
		return fmt.Errorf("marshal agent message: %w", err)
	}
	if len(data) > MaxMessageBytes {
		return fmt.Errorf("agent message exceeded %d bytes", MaxMessageBytes)
	}

	var header [4]byte
	binary.BigEndian.PutUint32(header[:], uint32(len(data)))

	e.mu.Lock()
	defer e.mu.Unlock()
	if _, err := e.w.Write(header[:]); err != nil {
		return err
	}
	_, err = e.w.Write(data)
	return err
}

type Decoder struct {
	r io.Reader
}

func NewDecoder(r io.Reader) *Decoder {
	return &Decoder{r: r}
}

func (d *Decoder) Decode() (*Message, error) {
	msg := &Message{}
	var header [4]byte
	if _, err := io.ReadFull(d.r, header[:]); err != nil {
		return msg, err
	}

	size := binary.BigEndian.Uint32(header[:])
	if size > MaxMessageBytes {
		return msg, fmt.Errorf("agent message exceeded %d bytes", MaxMessageBytes)
	}

	data := make([]byte, size)
	if _, err := io.ReadFull(d.r, data); err != nil {
		return msg, err
	}
	if err := proto.Unmarshal(data, msg); err != nil {
		return msg, fmt.Errorf("parse agent message: %w", err)
	}
	if err := validator.Validate(msg); err != nil {
		return msg, fmt.Errorf("validate agent message: %w", err)
	}
	return msg, nil
}