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) }