Function HandleStream

stdlib/httpsvr/handler.go:253–335

HandleStream wraps a StreamHandler into an http.HandlerFunc to handle streaming requests using Server-Sent Events (SSE).

(w http.ResponseWriter,
	r *http.Request, req Req, h StreamHandler[Req, Resp])

// HandleStream wraps a StreamHandler into an http.HandlerFunc to
// handle streaming requests using Server-Sent Events (SSE).
func HandleStream[Req RequestObject, Resp *Event[T], T any](w http.ResponseWriter,
	r *http.Request, req Req, h StreamHandler[Req, Resp]) {

	// Ensure the response writer supports flushing (required for SSE)
	flusher, ok := w.(http.Flusher)
	if !ok {
		err := errutil.Explain(nil, "streaming not supported")
		ErrorHandler(r, w, err)
		return
	}

	if err := ReadRequest(r, req); err != nil {
		ErrorHandler(r, w, err)
		return
	}

	done := make(chan struct{})
	responses := make(chan Resp)

	go func() {
		defer close(done)
		var res *Event[T]
		for res = range responses {

			select {
			case <-r.Context().Done():
				return
			default: // for linter
			}

			// Write SSE event
			if res.HasID() {
				if _, err := w.Write([]byte("id: " + res.GetID() + "\n")); err != nil {
					ErrorHandler(r, w, err)
					continue
				}
			}

			// Write SSE event
			if res.HasEvent() {
				if _, err := w.Write([]byte("event: " + res.GetEvent() + "\n")); err != nil {
					ErrorHandler(r, w, err)
					continue
				}
			}

			// Write SSE event
			if _, err := w.Write([]byte("data: ")); err != nil {
				ErrorHandler(r, w, err)
				continue
			}
			if err := jsonflow.MarshalWrite(w, res.GetData()); err != nil {
				ErrorHandler(r, w, err)
				continue
			}
			if _, err := w.Write([]byte("\n")); err != nil {
				ErrorHandler(r, w, err)
				continue