util

Documentation

Index

• type BufferableEvent
• type Element
  • func (e *Element[T]) Next() *Element[T]
  • func (e *Element[T]) Prev() *Element[T]
• type GroupTree
  • func NewGroupTree[T any]() *GroupTree[T]
  • func (g *GroupTree[T]) Add(interval adt.Interval, ch chan T)
  • func (g *GroupTree[T]) Broadcast(key adt.Interval, event T)
  • func (g *GroupTree[T]) Remove(interval adt.Interval, ch chan T)
• type List
  • func (l *List[T]) First() *Element[T]
  • func (l *List[T]) InsertAfter(v T, at *Element[T]) *Element[T]
  • func (l *List[T]) Last() *Element[T]
  • func (l *List[T]) PushBack(v T) *Element[T]
  • func (l *List[T]) PushFront(v T) *Element[T]
  • func (l *List[T]) Remove(e *Element[T]) T
• type TimeBuffer
  • func NewTimeBuffer[T any, TT BufferableEvent[T]](gapTimeout time.Duration, len int, ch chan<- TT) *TimeBuffer[T, TT]
  • func (t *TimeBuffer[T, TT]) All() (slice []TT)
  • func (t *TimeBuffer[T, TT]) LatestVisibleRev() int64
  • func (t *TimeBuffer[T, TT]) Len() int
  • func (t *TimeBuffer[T, TT]) Push(event TT)
  • func (t *TimeBuffer[T, TT]) Range(start int64, query T) ([]TT, int64, int64)
  • func (t *TimeBuffer[T, TT]) Run(ctx context.Context)

Types

type BufferableEvent

type BufferableEvent[T any] interface {
	GetAge() time.Duration
	GetRevision() int64
	Matches(query T) bool
}

type Element

type Element[T any] struct {
	Value T
	// contains filtered or unexported fields
}

func (*Element[T]) Next

func (e *Element[T]) Next() *Element[T]

func (*Element[T]) Prev

func (e *Element[T]) Prev() *Element[T]

type GroupTree

type GroupTree[T any] struct {
	// contains filtered or unexported fields
}

GroupTree extends etcd's ADT package by grouping listeners such that they can overlap. Otherwise ADT only supports a single listener per keyspace.

func NewGroupTree

func NewGroupTree[T any]() *GroupTree[T]

func (*GroupTree[T]) Add

func (g *GroupTree[T]) Add(interval adt.Interval, ch chan T)

func (*GroupTree[T]) Broadcast

func (g *GroupTree[T]) Broadcast(key adt.Interval, event T)

func (*GroupTree[T]) Remove

func (g *GroupTree[T]) Remove(interval adt.Interval, ch chan T)

type List

type List[T any] struct {
	Len int
	// contains filtered or unexported fields
}

List was adopted from https://gist.github.com/pje/90e727f80685c78a6c1cfff35f62155a. Replaec with container/list once it's generic.

func (*List[T]) First

func (l *List[T]) First() *Element[T]

func (*List[T]) InsertAfter

func (l *List[T]) InsertAfter(v T, at *Element[T]) *Element[T]

func (*List[T]) Last

func (l *List[T]) Last() *Element[T]

func (*List[T]) PushBack

func (l *List[T]) PushBack(v T) *Element[T]

func (*List[T]) PushFront

func (l *List[T]) PushFront(v T) *Element[T]

func (*List[T]) Remove

func (l *List[T]) Remove(e *Element[T]) T

type TimeBuffer

type TimeBuffer[T any, TT BufferableEvent[T]] struct {
	// contains filtered or unexported fields
}

TimeBuffer buffers events and sorts them by their logical timestamp.

Events are only visible to readers when the preceding event has been received. If the preceding event is never received, the following event will still become visible after a (wallclock) timeout period in order to prevent deadlocks caused dropped events.

In practice, it is useful for merging streams of events that may be out of order due to network latency, and have the possibility of missing events due to network partitions.

func NewTimeBuffer

func NewTimeBuffer[T any, TT BufferableEvent[T]](gapTimeout time.Duration, len int, ch chan<- TT) *TimeBuffer[T, TT]

func (*TimeBuffer[T, TT]) All

func (t *TimeBuffer[T, TT]) All() (slice []TT)

func (*TimeBuffer[T, TT]) LatestVisibleRev

func (t *TimeBuffer[T, TT]) LatestVisibleRev() int64

func (*TimeBuffer[T, TT]) Len

func (t *TimeBuffer[T, TT]) Len() int

func (*TimeBuffer[T, TT]) Push

func (t *TimeBuffer[T, TT]) Push(event TT)

func (*TimeBuffer[T, TT]) Range

func (t *TimeBuffer[T, TT]) Range(start int64, query T) ([]TT, int64, int64)

func (*TimeBuffer[T, TT]) Run

func (t *TimeBuffer[T, TT]) Run(ctx context.Context)