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