generic

Documentation

Index

• Constants
• type Broadcaster
  • func NewBroadcaster[T any](bufferSize int) *Broadcaster[T]
  • func (b *Broadcaster[T]) Broadcast(data T) error
  • func (b *Broadcaster[T]) Close()
  • func (b *Broadcaster[T]) IsClosed() bool
  • func (b *Broadcaster[T]) SetAutoclose(val bool)
  • func (b *Broadcaster[T]) Subscribe() (chan T, error)
  • func (b *Broadcaster[T]) Unsubscribe(ch chan T)
• type RingBuffer
  • func NewRingBuffer[T any](size int) *RingBuffer[T]
  • func (r *RingBuffer[T]) Dequeue() T
  • func (r *RingBuffer[T]) Drain() []T
  • func (r *RingBuffer[T]) Each(f func(any))
  • func (r *RingBuffer[T]) Enqueue(data T)

Constants

const (
	SafeWriteTimeout            = time.Duration(100) * time.Millisecond
	DefaultBroadcastChannelSize = 8
)

const DefaultRingBufferSize = 16384

Types

type Broadcaster

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

func NewBroadcaster

func NewBroadcaster[T any](bufferSize int) *Broadcaster[T]

func (*Broadcaster[T]) Broadcast

func (b *Broadcaster[T]) Broadcast(data T) error

Broadcast will send the provided T to all currently subscribed channels as long as the channel is not full (which would cause the broadcaster to block until there was space on that channel).

func (*Broadcaster[T]) Close

func (b *Broadcaster[T]) Close()

func (*Broadcaster[T]) IsClosed

func (b *Broadcaster[T]) IsClosed() bool

func (*Broadcaster[T]) SetAutoclose

func (b *Broadcaster[T]) SetAutoclose(val bool)

func (*Broadcaster[T]) Subscribe

func (b *Broadcaster[T]) Subscribe() (chan T, error)

Subscribe allows a client to request future entries and it does this by listening to the channel it is provided.

func (*Broadcaster[T]) Unsubscribe

func (b *Broadcaster[T]) Unsubscribe(ch chan T)

Unsubscribe will remove the provided channel (which should have come from a previous call to Subscribe)

type RingBuffer

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

func NewRingBuffer

func NewRingBuffer[T any](size int) *RingBuffer[T]

func (*RingBuffer[T]) Dequeue

func (r *RingBuffer[T]) Dequeue() T

func (*RingBuffer[T]) Drain

func (r *RingBuffer[T]) Drain() []T

func (*RingBuffer[T]) Each

func (r *RingBuffer[T]) Each(f func(any))

func (*RingBuffer[T]) Enqueue

func (r *RingBuffer[T]) Enqueue(data T)

Enqueue will continue to write data to the ring buffer even if nothing is reading from it.