Method Start

intercept/eventstream/eventstream.go:90–143

Start handles sending Server-Sent Event to the client.

(w http.ResponseWriter, r *http.Request)

// Start handles sending Server-Sent Event to the client.
func (s *EventStream) Start(w http.ResponseWriter, r *http.Request) {
	// Signal completion on exit so senders don't block indefinitely after closure.
	defer close(s.doneCh)

	ctx := r.Context()

	defer s.tick.Stop()

	for {
		var (
			ev   event
			open bool
		)

		select {
		case <-s.ctx.Done():
			return
		case <-ctx.Done():
			s.logger.Debug(ctx, "request context canceled", slog.Error(ctx.Err()))
			return
		case ev, open = <-s.eventsCh: // Once closed, the buffered channel will drain all buffered values before showing as closed.
			if !open {
				s.logger.Debug(ctx, "events channel closed")
				return
			}

			// Initiate the stream on first event (if not already initiated).
			s.InitiateStream(w)
		case <-s.tick.C:
			ev = s.pingPayload
			if ev == nil {
				continue
			}
		}

		_, err := w.Write(ev)
		if err != nil {
			if IsConnError(err) {
				s.logger.Debug(ctx, "client disconnected during SSE write", slog.Error(err))
			} else {
				s.logger.Warn(ctx, "failed to write SSE event", slog.Error(err))
			}
			return
		}
		if err := flush(w); err != nil {
			s.logger.Warn(ctx, "failed to flush event stream", slog.Error(err))
			return
		}

		// Reset the timer once we've flushed some data to the stream, since it's already fresh.
		// No need to ping in that case.
		s.tick.Reset(pingInterval)
	}
}

// Send enqueues an event in a non-blocking fashion, but if the channel is full
// then it will block.