ringfilter.go

package gbuf

import (
	"errors"
	"io"

	"github.com/zalgonoise/gio/v2"
)

// RingFilter is a buffer that is connected end-to-end, which allows continuous
// reads and writes where the caller configures a callback function to process
// all items on each write.
type RingFilter[T any] struct {
	RingBuffer[T]

	fn func([]T) error
}

// Write sets the contents of `p` to the buffer, in sequential order.
// The return value n is the length of p; err comes from the process func.
// If the index in the buffer has not been yet read, the entire unread
// buffer value is sent to the configured process function.
func (r *RingFilter[T]) Write(p []T) (n int, err error) {
	n, _ = r.RingBuffer.Write(p)

	if err = r.fn(p); err != nil {
		return n, err
	}

	return n, nil
}

// WriteItem writes the T `item` to the buffer in the next position.
// The returned error comes from the configured filter function. If the index
// in the buffer has not been yet read, it will be overwritten.
func (r *RingFilter[T]) WriteItem(item T) (err error) {
	idx := r.write
	_ = r.RingBuffer.WriteItem(item)

	return r.fn(r.items[idx : idx+1 : idx+1])
}

// WriteTo writes data to w until the buffer is drained or an error occurs.
// The return value n is the number of T items written; it always fits into an
// int, but it is int64 to match the gio.WriterTo interface. Any error
// encountered during the write operation is also returned.
func (r *RingFilter[T]) WriteTo(b gio.Writer[T]) (n int64, err error) {
	n, err = r.RingBuffer.WriteTo(b)
	if errors.Is(err, ErrRingBufferNegativeRead) {
		return n, ErrRingFilterNegativeRead
	}

	return n, err
}

// ReadFrom reads data from b until EOF and appends it to the buffer, cycling
// the buffer as needed. For each read chunk, the process function is called
// with the written slice.
// The return value n is the number of T items read. Any error except io.EOF
// encountered during the read is also returned.
func (r *RingFilter[T]) ReadFrom(b gio.Reader[T]) (n int64, err error) {
	r.canUnread = false

	// read the stream's items until the reader is depleted of any more data
	for {
		// read from r.write until the end of the RingFilter buffer
		num, readErr := b.Read(r.items[r.write:])

		if num < 0 {
			return n, ErrRingFilterNegativeRead
		}

		if num > 0 {
			start := r.write
			n += int64(num)
			r.advanceWrite(num)

			if errFilter := r.fn(r.items[start : start+num]); errFilter != nil {
				return n, errFilter
			}
		}

		if errors.Is(readErr, io.EOF) {
			return n, nil
		}

		if readErr != nil {
			return n, readErr
		}
	}
}

// UnreadItem unreads the last T item returned by the most recent successful
// read operation that read at least one T item.
func (r *RingFilter[T]) UnreadItem() error {
	if err := r.RingBuffer.UnreadItem(); err != nil {
		return ErrRingFilterUnreadItem
	}

	return nil
}

// Seek implements the gio.Seeker interface. All valid whence will point to
// the current cursor's (read) position.
func (r *RingFilter[T]) Seek(offset int64, whence int) (abs int64, err error) {
	abs, err = r.RingBuffer.Seek(offset, whence)
	if err != nil {
		return 0, ErrRingFilterInvalidWhence
	}

	return abs, nil
}

// NewRingFilter creates a RingFilter of type `T` and size `size`, with process function `fn`.
func NewRingFilter[T any](size int, fn func([]T) error) *RingFilter[T] {
	if size <= 0 {
		size = defaultBufferSize
	}

	if fn == nil {
		fn = unimplementedFilter[T]
	}

	return &RingFilter[T]{
		RingBuffer: RingBuffer[T]{
			items: make([]T, size),
		},
		fn: fn,
	}
}

func unimplementedFilter[T any]([]T) error {
	return nil
}