Repository: etcd-io/etcd
File: cache/ringbuffer.go (Go)

Code before:
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package cache

type ringBuffer[T any] struct {
	buffer []entry[T]
	// head is the index immediately after the last non-empty entry in the buffer (i.e., the next write position).
	head, tail, entryCount int
}

type entry[T any] struct {
	revision int64
	items    []T
}

type KeyPredicate = func([]byte) bool

func newRingBuffer[T any](capacity int) *ringBuffer[T] {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer[T]{
		buffer: make([]entry[T], capacity),
	}
}

type RevisionOf[T any] func(T) int64

func (r *ringBuffer[T]) Append(items []T, revOf RevisionOf[T]) {
	start := 0
	for end := 1; end < len(items); end++ {
		if revOf(items[end]) != revOf(items[start]) {
			r.append(entry[T]{
				revision: revOf(items[start]),
				items:    items[start:end],
			})
			start = end
		}
	}
	if start < len(items) {
		r.append(entry[T]{
			revision: revOf(items[start]),
			items:    items[start:],

Review comment (suggestion):
Maybe instead:
* entry having `items []T`
* demux storing `history *ringBuffer[*clientv3.Event]`

We could implement:
* entry having `item T`
* demux storing `history *ringBuffer[[]*clientv3.Event]`

Code after:
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
//     http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.

package cache

import "fmt"

type ringBuffer[T any] struct {
	buffer []entry[T]
	// head is the index immediately after the last non-empty entry in the buffer (i.e., the next write position).
	head, tail, size int
	revisionOf       RevisionOf[T]
}

type entry[T any] struct {
	revision int64
	items    []T
}

type (
	KeyPredicate      = func([]byte) bool
	RevisionOf[T any] func(T) int64
	IterFunc[T any]   func(rev int64, items []T) bool
)

func newRingBuffer[T any](capacity int, revisionOf RevisionOf[T]) *ringBuffer[T] {
	// assume capacity > 0 – validated by Cache
	return &ringBuffer[T]{
		buffer:     make([]entry[T], capacity),
		revisionOf: revisionOf,
	}
}

func (r *ringBuffer[T]) Append(items []T) {
	start := 0
	for end := 1; end < len(items); end++ {
		if r.revisionOf(items[end]) != r.revisionOf(items[start]) {
			r.append(entry[T]{
				revision: r.revisionOf(items[start]),
				items:    items[start:end],
			})
			start = end