qb/txlog/txlog.go

package txlog

import (
	"io"

	bin "gordenko.dev/dima/bin/little"
	"gordenko.dev/dima/qb"
	"gordenko.dev/dima/qb/atree"
	"gordenko.dev/dima/qb/conbuf"
	"gordenko.dev/dima/qb/proto"
)

type AddedMetric struct {
	MetricID   uint32
	MetricType qb.MetricType
	FracDigits int
}

func (s AddedMetric) Pack(w io.Writer) {
	arr := []byte{
		CodeAddedMetric,
		0, 0, 0, 0, //
		byte(s.MetricType),
		byte(s.FracDigits),
	}
	bin.PutUint32(arr[1:], s.MetricID)
	w.Write(arr)
}

func (s *AddedMetric) Parse(src io.Reader) (err error) {
	arr, err := bin.ReadN(src, 6)
	if err != nil {
		return
	}
	s.MetricID, _ = bin.GetUint32(arr)
	s.MetricType = qb.MetricType(arr[4])
	s.FracDigits = int(arr[5])
	return nil
}

type DeletedMetric struct {
	MetricID        uint32
	FreePageNumbers []uint32
}

func (s DeletedMetric) Pack(w io.Writer) {
	arr := []byte{
		CodeDeletedMetric,
		0, 0, 0, 0, // metricID
	}
	bin.PutUint32(arr[1:], s.MetricID)
	w.Write(arr)
	bin.WriteVarSize(w, len(s.FreePageNumbers))
	for _, pageNo := range s.FreePageNumbers {
		bin.WriteUint32(w, pageNo)
	}
}

func (s *DeletedMetric) Parse(src io.Reader) (err error) {
	s.MetricID, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	// free pages
	qty, err := bin.ReadVarSize(src)
	if err != nil {
		return
	}
	for range qty {
		var pageNo uint32
		pageNo, err = bin.ReadUint32(src)
		if err != nil {
			return
		}
		s.FreePageNumbers = append(s.FreePageNumbers, pageNo)
	}
	return nil
}

type AppendedMeasure struct {
	MetricID  uint32
	Timestamp uint32
	Value     float64
}

func (s AppendedMeasure) Pack(w io.Writer) {
	arr := []byte{
		CodeAppendedMeasure,
		0, 0, 0, 0, // metricID
		0, 0, 0, 0, // timestamp
		0, 0, 0, 0, 0, 0, 0, 0, // value
	}
	bin.PutUint32(arr[1:], s.MetricID)
	bin.PutUint32(arr[5:], s.Timestamp)
	bin.PutFloat64(arr[9:], s.Value)
	w.Write(arr)
}

func (s *AppendedMeasure) Parse(src io.Reader) (err error) {
	arr, err := bin.ReadN(src, 16)
	if err != nil {
		return
	}
	s.MetricID, _ = bin.GetUint32(arr[0:])
	s.Timestamp, _ = bin.GetUint32(arr[4:])
	s.Value, _ = bin.GetFloat64(arr[8:])
	return nil
}

type AppendedMeasures struct {
	MetricID uint32
	Measures []proto.Measure
}

func (s AppendedMeasures) Pack(w io.Writer) {
	arr := []byte{
		CodeAppendedMeasures,
		0, 0, 0, 0, // metricID
		0, 0, // qty
	}
	bin.PutUint32(arr[1:], s.MetricID)
	bin.PutUint16(arr[5:], uint16(len(s.Measures)))
	w.Write(arr)
	for _, measure := range s.Measures {
		bin.WriteUint32(w, measure.Timestamp)
		bin.WriteFloat64(w, measure.Value)
	}
}

func (s *AppendedMeasures) Parse(src io.Reader) (err error) {
	s.MetricID, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	qty, err := bin.ReadUint16(src)
	if err != nil {
		return
	}
	for range qty {
		var measure proto.Measure
		measure.Timestamp, err = bin.ReadUint32(src)
		if err != nil {
			return
		}
		measure.Value, err = bin.ReadFloat64(src)
		if err != nil {
			return
		}
		s.Measures = append(s.Measures, measure)
	}
	return nil
}

type AppendedPagesReq struct {
	Legs      []atree.PathLeg
	MetricID  uint32
	Timestamp uint32  // last measure
	Value     float64 // last  measure
	//RootPageNo       uint32
	LastPageNo       uint32
	Pages            []atree.NotLinkedDataPage
	TimestampsChunks [][]byte
	TimestampsSize   uint16
	ValuesChunks     [][]byte
	ValuesSize       uint16
}

/*
4b   metricID
4b   timestamp
8b   value
4b   LastPageNo (0 if no new root page)
4b   newRootPageNo (0 if no new root page)
Nb - varsize (pages qty)
[

	4b  pageNo
	1b  isReused
	Xb  page (page size)

]
2b - timestamps size (not filled page)
Nb - timestamps payload
2b - values size (not filled page)
Nb - values payload
*/
type AppendedPages struct {
	MetricID         uint32
	Timestamp        uint32  // last measure
	Value            float64 // last  measure
	NewRootPageNo    uint32
	LastPageNo       uint32
	Pages            []atree.PageToWrite
	TimestampsChunks [][]byte
	TimestampsSize   int
	ValuesChunks     [][]byte
	ValuesSize       int
}

func (s *AppendedPages) Pack(w io.Writer) {
	w.Write([]byte{
		CodeAppendedPages,
	})
	bin.WriteUint32(w, s.MetricID)
	bin.WriteUint32(w, s.Timestamp)
	bin.WriteFloat64(w, s.Value)
	bin.WriteUint32(w, s.LastPageNo)
	bin.WriteUint32(w, s.NewRootPageNo)
	//
	bin.WriteVarSize(w, len(s.Pages))
	for _, p := range s.Pages {
		bin.WriteUint32(w, p.PageNo)
		if p.IsReused {
			w.Write([]byte{1})
		} else {
			w.Write([]byte{0})
		}
		w.Write(p.Data)
	}
	// timestamps
	writeChunkedPayloadTo(s.TimestampsChunks, int(s.TimestampsSize), w)
	// values
	writeChunkedPayloadTo(s.ValuesChunks, int(s.ValuesSize), w)
}

func (s *AppendedPages) Parse(src io.Reader) (err error) {
	s.MetricID, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	s.Timestamp, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	s.Value, err = bin.ReadFloat64(src)
	if err != nil {
		return
	}
	s.LastPageNo, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	s.NewRootPageNo, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	pagesQty, err := bin.ReadVarSize(src)
	if err != nil {
		return
	}
	for range pagesQty {
		var page atree.PageToWrite
		page.PageNo, err = bin.ReadUint32(src)
		if err != nil {
			return
		}
		page.IsReused, err = bin.ReadBool(src)
		if err != nil {
			return
		}
		page.Data, err = bin.ReadN(src, atree.PageSize)
		if err != nil {
			return
		}
		s.Pages = append(s.Pages, page)
	}
	s.TimestampsChunks, s.TimestampsSize, err = readChunkedPayload(src)
	if err != nil {
		return
	}
	s.ValuesChunks, s.ValuesSize, err = readChunkedPayload(src)
	if err != nil {
		return
	}
	return nil
}

func writeChunkedPayloadTo(chunks [][]byte, size int, w io.Writer) {
	bin.WriteVarSize(w, size)
	for _, chunk := range chunks {
		if size >= len(chunk) {
			w.Write(chunk)
			size -= len(chunk)
		} else {
			w.Write(chunk[:size])
			return
		}
	}
}

func readChunkedPayload(src io.Reader) (_ [][]byte, _ int, err error) {
	var (
		chunks        [][]byte
		size          int
		remainingSize = size
	)
	size, err = bin.ReadVarSize(src)
	if err != nil {
		return
	}
	for remainingSize > 0 {
		var chunk []byte
		if remainingSize >= conbuf.ChunkSize {
			chunk, err = bin.ReadN(src, conbuf.ChunkSize)
			if err != nil {
				return
			}
		} else {
			chunk = make([]byte, conbuf.ChunkSize)
			err = bin.ReadNInto(src, chunk[:remainingSize])
			if err != nil {
				return
			}
		}
		chunks = append(chunks, chunk)
		remainingSize -= conbuf.ChunkSize
	}
	return chunks, size, nil
}

// fix - add changed pages
type DeletedMeasures struct {
	MetricID        uint32
	FreePageNumbers []uint32
}

func (s DeletedMeasures) Pack(w io.Writer) {
	arr := []byte{
		CodeDeletedMeasures,
		0, 0, 0, 0,
	}
	bin.PutUint32(arr[1:], s.MetricID)
	w.Write(arr)
	bin.WriteVarSize(w, len(s.FreePageNumbers))
	for _, pageNo := range s.FreePageNumbers {
		bin.WriteUint32(w, pageNo)
	}
}

func (s *DeletedMeasures) Parse(src io.Reader) (err error) {
	s.MetricID, err = bin.ReadUint32(src)
	if err != nil {
		return
	}
	// free pages
	qty, err := bin.ReadVarSize(src)
	if err != nil {
		return
	}
	for range qty {
		var pageNo uint32
		pageNo, err = bin.ReadUint32(src)
		if err != nil {
			return
		}
		s.FreePageNumbers = append(s.FreePageNumbers, pageNo)
	}
	return nil
}