1- // Package sse provides Server-Sent Events (SSE) endpoint for streaming log files.
21package sse
32
43import (
@@ -20,8 +19,8 @@ const (
2019 sseDoneEvent = "done" // Event name for the end of the stream.
2120)
2221
23- // GetSSEHandler creates and returns an http.HandlerFunc for the SSE endpoint.
24- // It takes the application's logger as an argument .
22+ // GetSSEHandler creates and returns an
23+ // http.HandlerFunc for the SSE endpoint .
2524func GetSSEHandler (logger util.Logger ) http.HandlerFunc {
2625 return func (w http.ResponseWriter , r * http.Request ) {
2726 serveSSE (logger , w , r )
@@ -30,19 +29,17 @@ func GetSSEHandler(logger util.Logger) http.HandlerFunc {
3029
3130// serveSSE handles incoming SSE requests.
3231func serveSSE (logger util.Logger , w http.ResponseWriter , r * http.Request ) {
33- ctx := r .Context () // Get context from the request
32+ ctx := r .Context ()
3433
35- // --- 1. Get Run ID from Header ---
34+ // Get Run ID from Header.
3635 runId := r .Header .Get (runIdHeader )
3736 if runId == "" {
3837 logger .Warn (fmt .Sprintf ("[SSE Handler] Missing %s header" , runIdHeader ))
3938 http .Error (w , fmt .Sprintf ("Missing %s header" , runIdHeader ), http .StatusBadRequest )
4039 return
4140 }
4241
43- // --- Basic Sanitize Run ID (Prevent path traversal) ---
44- // Ensure runId doesn't contain potentially harmful sequences.
45- // A more robust solution might involve checking against a list of valid run IDs.
42+ // Basic Sanitize Run ID (Prevent path traversal)
4643 if strings .Contains (runId , ".." ) || strings .ContainsAny (runId , "/\\ " ) {
4744 logger .Warn (fmt .Sprintf ("[SSE Handler] Invalid characters in %s header: %s" , runIdHeader , runId ))
4845 http .Error (w , "Invalid Run ID format" , http .StatusBadRequest )
@@ -54,16 +51,15 @@ func serveSSE(logger util.Logger, w http.ResponseWriter, r *http.Request) {
5451
5552 logger .Info (fmt .Sprintf ("[SSE Handler] Request received for runId: %s (File: %s)" , runId , logFilePath ))
5653
57- // --- 2. Set SSE Headers ---
54+ // Set SSE Headers.
5855 w .Header ().Set ("Content-Type" , "text/event-stream" )
5956 w .Header ().Set ("Cache-Control" , "no-cache" )
6057 w .Header ().Set ("Connection" , "keep-alive" )
6158 w .Header ().Set ("X-Accel-Buffering" , "no" )
6259
63- // Suggest a retry interval to the client
64- fmt .Fprintf (w , "retry: %d \n \n " , retrySeconds * 1000 ) // retry is in milliseconds
60+ // Suggest a retry interval to the client.
61+ fmt .Fprintf (w , "retry: %ds \n \n " , retrySeconds )
6562
66- // --- 3. Get Flusher ---
6763 rc := http .NewResponseController (w )
6864 if rc == nil {
6965 logger .Error (fmt .Sprintf ("[SSE Handler] Failed to get ResponseController for runId: %s" , runId ))
@@ -77,14 +73,13 @@ func serveSSE(logger util.Logger, w http.ResponseWriter, r *http.Request) {
7773 return
7874 }
7975
80- // --- 4. Configure and Start Tailing ---
76+ // Configure and Start Tailing.
8177 tailConfig := tail.Config {
82- Location : & tail.SeekInfo {Offset : 0 , Whence : io .SeekStart }, // Start from the beginning of the file
83- ReOpen : true , // Re-open file if recreated (log rotation)
84- MustExist : false , // Don't fail if file doesn't exist yet
85- Poll : true , // Use polling (more reliable across FS types/network mounts than pure inotify) - tune if needed
86- Follow : true , // Keep following the file for new lines
87- // Logger: tail.DiscardingLogger, // Uncomment to disable tail library's internal logging
78+ Location : & tail.SeekInfo {Offset : 0 , Whence : io .SeekStart },
79+ ReOpen : true ,
80+ MustExist : false ,
81+ Poll : true ,
82+ Follow : true ,
8883 }
8984
9085 tailer , err := tail .TailFile (logFilePath , tailConfig )
@@ -94,70 +89,73 @@ func serveSSE(logger util.Logger, w http.ResponseWriter, r *http.Request) {
9489 return
9590 }
9691
97- // Ensure tailer is stopped when the handler exits
92+ // Ensure tailer is stopped when the handler exits.
9893 defer func () {
9994 logger .Info (fmt .Sprintf ("[SSE Handler] Stopping tailer for runId: %s" , runId ))
100- // Stopping the tailer closes its internal channels
95+ // Stopping the tailer closes its internal channels.
10196 if stopErr := tailer .Stop (); stopErr != nil {
10297 logger .Error (fmt .Sprintf ("[SSE Tailing] Error stopping tailer for runId %s: %v" , runId , stopErr ))
10398 }
10499 }()
105100
106101 logger .Info (fmt .Sprintf ("[SSE Handler] Started tailing %s for runId: %s" , logFilePath , runId ))
107102
108- // --- 5. Event Loop: Send lines and handle disconnect ---
103+ // Event Loop - Send lines and handle disconnect.
109104 for {
110105 select {
111- case <- ctx .Done (): // Client disconnected
106+ case <- ctx .Done ():
107+ // Client disconnected.
112108 logger .Info (fmt .Sprintf ("[SSE Handler] Client disconnected for runId: %s" , runId ))
113- return // Exit handler, defer will stop tailer
109+ return
114110
115- case line , ok := <- tailer .Lines : // New line from file
111+ case line , ok := <- tailer .Lines :
116112 if ! ok {
117- // Channel closed, tailer might have stopped or encountered an error
113+ // Channel closed, tailer might
114+ // have stopped or encountered an error.
118115 tailErr := tailer .Err ()
119- if tailErr != nil && tailErr != io .EOF { // io.EOF might occur normally on stop
116+ if tailErr != nil && tailErr != io .EOF {
120117 logger .Error (fmt .Sprintf ("[SSE Tailing] Error during tailing for runId %s: %v" , runId , tailErr ))
121118 } else {
122119 logger .Info (fmt .Sprintf ("[SSE Tailing] Tailer lines channel closed for runId: %s" , runId ))
123120 }
124- return // Exit handler
121+ return
125122 }
126123
127- fmt .Printf ("[SSE Handler] New line for runId %s: %s\n " , runId , line .Text ) // Debug log
124+ // Debug log.
125+ fmt .Printf ("[SSE Handler] New line for runId %s: %s\n " , runId , line .Text )
128126
129- // ---> CHECK FOR END MARKER <---
130127 if line .Text == "__END__" {
131128 logger .Info (fmt .Sprintf ("[SSE Handler] Detected END marker for run %s. Sending '%s' event and closing stream." , runId , sseDoneEvent ))
132- // Send the specific "done" event
129+
130+ // Send SSE event indicating stream end.
133131 _ , writeErr := fmt .Fprintf (w , "event: %s\n data: {\" message\" : \" Stream ended.\" }\n \n " , sseDoneEvent )
134132 if writeErr != nil {
135- // Log error but still attempt to flush and return
136133 logger .Warn (fmt .Sprintf ("[SSE Handler] Error writing '%s' event for runId %s: %v" , sseDoneEvent , runId , writeErr ))
137134 }
138- // Attempt to flush the final event
135+
136+ // Attempt to flush the final event.
139137 if err := rc .Flush (); err != nil {
140138 logger .Error (fmt .Sprintf ("[SSE Handler] Error flushing final event for runId %s: %v" , runId , err ))
141- return // Exit handler, defer will stop tailer
139+ return
142140 }
143- return // <<< EXIT HANDLER HERE after sending event
141+ return
144142 }
145143
146- // Format and send SSE message
147- // Use fmt.Fprintf to write directly to the ResponseWriter
148- _ , writeErr := fmt .Fprintf (w , "data: %s\n \n " , line .Text ) // SSE format: "data: content\n\n"
144+ // Format and send SSE message.
145+ // SSE format: "data: content\n\n"
146+ _ , writeErr := fmt .Fprintf (w , "data: %s\n \n " , line .Text )
149147
150148 if writeErr != nil {
151- // Likely client disconnected or network error
152149 logger .Warn (fmt .Sprintf ("[SSE Handler] Error writing to client for runId %s: %v" , runId , writeErr ))
153- return // Exit handler, defer will stop tailer
150+ return
154151 }
155152
156153 if err := rc .Flush (); err != nil {
157154 logger .Error (fmt .Sprintf ("[SSE Handler] Error flushing response for runId %s: %v" , runId , err ))
158- return // Exit handler, defer will stop tailer
155+ return
159156 }
160157
158+ // Sleep to prevent overwhelming the client.
161159 time .Sleep (50 * time .Millisecond )
162160 }
163161 }
0 commit comments