67 lines
1.3 KiB
Go
67 lines
1.3 KiB
Go
package shell
|
|
|
|
import "sync"
|
|
|
|
// Subscription is a consumer handle for a shell output stream.
|
|
type Subscription interface {
|
|
Output() <-chan []byte
|
|
Close() error
|
|
}
|
|
|
|
// subscription is a buffered, closable fan-out channel for a Session.
|
|
type subscription struct {
|
|
session *Session
|
|
ch chan []byte
|
|
once sync.Once
|
|
mu sync.Mutex
|
|
closed bool
|
|
}
|
|
|
|
func newSubscription(sess *Session) *subscription {
|
|
return &subscription{
|
|
session: sess,
|
|
ch: make(chan []byte, 32),
|
|
}
|
|
}
|
|
|
|
func (s *subscription) Output() <-chan []byte {
|
|
return s.ch
|
|
}
|
|
|
|
func (s *subscription) Close() error {
|
|
s.once.Do(func() {
|
|
s.mu.Lock()
|
|
s.closed = true
|
|
close(s.ch)
|
|
s.mu.Unlock()
|
|
s.session.removeSubscription(s)
|
|
})
|
|
return nil
|
|
}
|
|
|
|
// closeChan closes the underlying channel exactly once without removing the
|
|
// subscription from the session's subscriber list. It is used by the manager
|
|
// when the shell exits.
|
|
func (s *subscription) closeChan() {
|
|
s.once.Do(func() {
|
|
s.mu.Lock()
|
|
s.closed = true
|
|
close(s.ch)
|
|
s.mu.Unlock()
|
|
})
|
|
}
|
|
|
|
// send performs a non-blocking send to the subscription channel. If the
|
|
// channel is full or already closed, the chunk is dropped.
|
|
func (s *subscription) send(chunk []byte) {
|
|
s.mu.Lock()
|
|
defer s.mu.Unlock()
|
|
if s.closed {
|
|
return
|
|
}
|
|
select {
|
|
case s.ch <- chunk:
|
|
default:
|
|
}
|
|
}
|