Start handles sending Server-Sent Event to the client.
(w http.ResponseWriter, r *http.Request)
| 96 | |
| 97 | // Start handles sending Server-Sent Event to the client. |
| 98 | func (s *EventStream) Start(w http.ResponseWriter, r *http.Request) { |
| 99 | // Signal completion on exit so senders don't block indefinitely after closure. |
| 100 | defer close(s.doneCh) |
| 101 | |
| 102 | ctx := r.Context() |
| 103 | |
| 104 | defer s.tick.Stop() |
| 105 | |
| 106 | for { |
| 107 | var ( |
| 108 | ev event |
| 109 | open bool |
| 110 | ) |
| 111 | |
| 112 | select { |
| 113 | case <-s.ctx.Done(): |
| 114 | return |
| 115 | case <-ctx.Done(): |
| 116 | s.logger.Debug(ctx, "request context canceled", slog.Error(ctx.Err())) |
| 117 | return |
| 118 | case ev, open = <-s.eventsCh: // Once closed, the buffered channel will drain all buffered values before showing as closed. |
| 119 | if !open { |
| 120 | s.logger.Debug(ctx, "events channel closed") |
| 121 | return |
| 122 | } |
| 123 | |
| 124 | // Initiate the stream on first event (if not already initiated). |
| 125 | s.InitiateStream(w) |
| 126 | case <-s.tick.C: |
| 127 | ev = s.pingPayload |
| 128 | if ev == nil { |
| 129 | continue |
| 130 | } |
| 131 | } |
| 132 | |
| 133 | _, err := w.Write(ev) |
| 134 | if err != nil { |
| 135 | if IsConnError(err) { |
| 136 | s.logger.Debug(ctx, "client disconnected during SSE write", slog.Error(err)) |
| 137 | } else { |
| 138 | s.logger.Warn(ctx, "failed to write SSE event", slog.Error(err)) |
| 139 | } |
| 140 | return |
| 141 | } |
| 142 | flushStart := s.clk.Now() |
| 143 | if err := flush(w); err != nil { |
| 144 | s.logger.Warn(ctx, "failed to flush event stream", slog.Error(err)) |
| 145 | return |
| 146 | } |
| 147 | if d := s.clk.Since(flushStart); d > SlowFlushThreshold { |
| 148 | clientIP, _, _ := net.SplitHostPort(r.RemoteAddr) |
| 149 | s.logger.Warn(ctx, "slow client detected", |
| 150 | slog.F("flush_duration", d), |
| 151 | slog.F("client_ip", clientIP), |
| 152 | slog.F("user_agent", r.Header.Get("User-Agent")), |
| 153 | slog.F("payload_size", len(ev)), |
| 154 | ) |
| 155 | } |