delimbuf.go

// delimbuf is a byte buffer that tracks the "records" of data
// that is appended to it.
type delimbuf struct {
	buf []byte

	// Record offsets in the buffer are tracked as offsets relative
	// to base. The purpose of this is to allow Shift() to quickly
	// remove records without an O(N) operation to rebase the offsets
	// stored in delims based on the shifted buffer contents.
	//
	// Particularly obnoxious readers will notice that there is no
	// handling of wrap-arounds, which will almost certainly panic.
	// This would require pushing 9.6EiB through the buffer, which
	// is a little far-fetched. Perhaps my attitude would be different
	// if this was avionics and not a pipeline of JSON serdes.
	delims []int64
	base   int64
}

func (d *delimbuf) unbase(bo int64) int {
	return int(bo - d.base)
}

func (d *delimbuf) rebase(offset int) int64 {
	return int64(offset) + d.base
}

func (d *delimbuf) recordLen(index int) int {
	this := d.unbase(d.delims[index])
	next := len(d.buf)
	lastidx := len(d.delims) - 1
	if index < lastidx {
		next = d.unbase(d.delims[index+1])
	}
	return next - this
}

// Append adds record src to the buffer
func (d *delimbuf) Append(src []byte) {
	if d.buf == nil {
		d.buf = make([]byte, 0, len(src))
	}
	if d.delims == nil {
		d.delims = make([]int64, 0, 1)
	}

	offset := d.rebase(len(d.buf))
	d.delims = append(d.delims, offset)
	d.buf = append(d.buf, src...)
}

// BufferLen returns the number of bytes in the buffer
func (d *delimbuf) BufferLen() int {
	return len(d.buf)
}

// Len returns the number of buffered records
func (d *delimbuf) Len() int {
	return len(d.delims)
}

// Shift removes count records from front of the buffer
func (d *delimbuf) Shift(count int) {
	if len(d.buf) < count || len(d.delims) < count {
		panic("shift exceeds available elements")
	}

	idx := count - 1
	nbytes := d.unbase(d.delims[idx]) + d.recordLen(idx)
	d.delims = d.delims[count:]

	if nbytes > 0 {
		d.base += int64(nbytes)
		d.buf = d.buf[nbytes:]
	}
}

// Get returns a slice of the buffer for record at index
func (d *delimbuf) Get(index int) []byte {
	if d.buf == nil || d.delims == nil {
		panic("index out of bounds")
	}

	from := d.unbase(d.delims[index])
	to := from + d.recordLen(index)
	return d.buf[from:to]
}

// IndexAtOffset returns the record index for the buffer position
// specified by offset. This is an ~O(log n) operation where n is
// the number of records in the buffer.
func (d *delimbuf) IndexForOffset(offset int) int {
	if offset >= len(d.buf) {
		panic("offset is beyond buffer")
	}

	// Do a right-most binary search to find the record, taking advantage
	// of the monotonic nature of delims.
	left := 0
	right := len(d.delims)
	for left < right {
		mid := (left + right) / 2 // remainder is truncated
		if d.unbase(d.delims[mid]) > offset {
			right = mid
		} else {
			left = mid + 1
		}
	}
	return right - 1
}