goutil/sync/gate.go

package usync

import "sync"

// Error defines errors that this package can produce
type Error string; const (
	ErrAlreadyClosed Error = "AlreadyClosed"
)

// Error fullfills the error interface.
func (err Error) Error () string {
	return string(err)
}

// Gate wraps a channel and allows the receiver to abruptly stop receiving
// messages without causing the sender to lock up.
type Gate[T any] struct {
	channel    chan T
	lock       sync.RWMutex
	open       bool
	bufferSize int
}

// NewGate creates a new gate with no buffer.
func NewGate[T any] () Gate[T] {
	return Gate[T] {
		channel: make(chan T),
		open:    true,
	}
}

// NewBufferedGate creates a new gate with a buffer.
func NewBufferedGate[T any] (buffer int) Gate[T] {
	return Gate[T] {
		channel:    make(chan T, buffer),
		open:       true,
		bufferSize: buffer,
	}
}

// Send sends and item to the channel, returning whether the item was sent.
func (this *Gate[T]) Send (item T) bool {
	this.lock.RLock()
	defer this.lock.RUnlock()
	if !this.open { return false }
	this.channel <- item
	return true
}

// Receive returns a receive-only channel that can be used to receive items.
func (this *Gate[T]) Receive () <- chan T {
	this.lock.RLock()
	defer this.lock.RUnlock()
	return this.channel
}

// Close closes the gate, drains all remaining messages, and closes the channel.
func (this *Gate[T]) Close () error {
	this.lock.Lock()
	if !this.open { return ErrAlreadyClosed }
	this.open = false
	this.lock.Unlock()
	for len(this.channel) > 0 { <- this.channel }
	close(this.channel)
	return nil
}

// Reset re-opens the gate if it is closed, and creates a new channel.
func (this *Gate[T]) Reset () error {
	this.lock.Lock()
	defer this.lock.Unlock()
	this.open = true
	if this.channel != nil { close(this.channel) }
	this.channel = make(chan T, this.bufferSize)
	return nil
}

// Open returns whether the gate is open.
func (this *Gate[T]) Open () bool {
	this.lock.RLock()
	defer this.lock.RUnlock()
	return this.open
}

// Len returns the amount of items in the channel.
func (this *Gate[T]) Len () int {
	return len(this.channel)
}

// Cap returns the amount of items the channel can hold.
func (this *Gate[T]) Cap () int {
	return cap(this.channel)
}