Method Shutdown

aibridge/intercept/eventstream/eventstream.go:195–218

Shutdown gracefully shuts down the stream, sending any supplementary events downstream if required. ONLY call this once all events have been submitted.

(shutdownCtx context.Context)

// Shutdown gracefully shuts down the stream, sending any supplementary events downstream if required.
// ONLY call this once all events have been submitted.
func (s *EventStream) Shutdown(shutdownCtx context.Context) error {
	s.shutdownOnce.Do(func() {
		s.logger.Debug(shutdownCtx, "shutdown initiated", slog.F("outstanding_events", len(s.eventsCh)))

		// Now it is safe to close the events channel; the Start() loop will exit
		// after draining remaining events and receivers will stop ranging.
		close(s.eventsCh)
	})

	var err error
	select {
	case <-shutdownCtx.Done():
		// If shutdownCtx completes, shutdown likely exceeded its timeout.
		err = xerrors.Errorf("shutdown ended prematurely with %d outstanding events: %w", len(s.eventsCh), shutdownCtx.Err())
	case <-s.ctx.Done():
		err = xerrors.Errorf("shutdown ended prematurely with %d outstanding events: %w", len(s.eventsCh), s.ctx.Err())
	case <-s.doneCh:
		return nil
	}

	// Even if the context is canceled, we need to wait for Start() to complete.
	<-s.doneCh
	return err
}

// IsStreaming checks if the stream has been initiated, or
// when events are buffered which - when processed - will initiate the stream.