sync: Don't panic on double close of Gate, return error instead

sync/gate.go

@@ -2,12 +2,23 @@ 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.
@@ -23,24 +34,30 @@ 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 {
-	if !this.Open() { return false }
+	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 }
@@ -48,6 +65,16 @@ func (this *Gate[T]) Close () error {
 	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()