5887ab5ddb
* feat(tracker): update message schema with BatchMetadata; separate message-related responsibilities; add message size * chore(docker): removing edge busybox, as the main repo updated * feat(backend): updated message protocol templates * feat(backend): added support of message size * feat(backend): implemented iterator for new message protocol (with message size) Co-authored-by: Alex Kaminskii <alex@openreplay.com>
69 lines
1.2 KiB
Go
69 lines
1.2 KiB
Go
package messages
|
|
|
|
import (
|
|
"bytes"
|
|
"encoding/binary"
|
|
"io"
|
|
"log"
|
|
)
|
|
|
|
// RawMessage is a not decoded message
|
|
type RawMessage struct {
|
|
tp uint64
|
|
size uint64
|
|
data []byte
|
|
reader *bytes.Reader
|
|
meta *message
|
|
encoded bool
|
|
skipped *bool
|
|
}
|
|
|
|
func (m *RawMessage) Encode() []byte {
|
|
if m.encoded {
|
|
return m.data
|
|
}
|
|
m.data = make([]byte, m.size+1)
|
|
m.data[0] = uint8(m.tp)
|
|
m.encoded = true
|
|
*m.skipped = false
|
|
n, err := io.ReadFull(m.reader, m.data[1:])
|
|
if err != nil {
|
|
log.Printf("message encode err: %s", err)
|
|
return nil
|
|
}
|
|
log.Printf("encode: read %d of %d bytes", n, m.size)
|
|
return m.data
|
|
}
|
|
|
|
func (m *RawMessage) EncodeWithIndex() []byte {
|
|
if !m.encoded {
|
|
m.Encode()
|
|
}
|
|
if IsIOSType(int(m.tp)) {
|
|
return m.data
|
|
}
|
|
data := make([]byte, len(m.data)+8)
|
|
copy(data[8:], m.data[:])
|
|
binary.LittleEndian.PutUint64(data[0:], m.Meta().Index)
|
|
return data
|
|
}
|
|
|
|
func (m *RawMessage) Decode() Message {
|
|
if !m.encoded {
|
|
m.Encode()
|
|
}
|
|
msg, err := ReadMessage(m.tp, bytes.NewReader(m.data[1:]))
|
|
if err != nil {
|
|
log.Printf("decode err: %s", err)
|
|
}
|
|
msg.Meta().SetMeta(m.meta)
|
|
return msg
|
|
}
|
|
|
|
func (m *RawMessage) TypeID() int {
|
|
return int(m.tp)
|
|
}
|
|
|
|
func (m *RawMessage) Meta() *message {
|
|
return m.meta
|
|
}
|