diff --git a/server/ai_http.go b/server/ai_http.go index edb35664e..c2937b2ce 100644 --- a/server/ai_http.go +++ b/server/ai_http.go @@ -73,9 +73,6 @@ func startAIServer(lp *lphttp) error { lp.transRPC.Handle("/live-video-to-video", oapiReqValidator(lp.StartLiveVideoToVideo())) // Additionally, there is the '/aiResults' endpoint registered in server/rpc.go - // This endpoint is used to get the latest status of a live-video-to-video stream - lp.transRPC.HandleFunc("/stream-status/{streamID}", lp.handleStreamStatus()) - return nil } @@ -804,33 +801,3 @@ func parseMultiPartResult(body io.Reader, boundary string, pipeline string) core return wkrResult } - -// handleStreamStatus returns the latest available status of a live-video-to-video stream -func (h *lphttp) handleStreamStatus() http.HandlerFunc { - return func(w http.ResponseWriter, r *http.Request) { - if r.Method != http.MethodGet { - http.Error(w, "Method not allowed", http.StatusMethodNotAllowed) - return - } - - streamID := strings.TrimPrefix(r.URL.Path, "/stream-status/") - if streamID == "" { - respondWithError(w, "stream ID is required", http.StatusBadRequest) - return - } - - // Get status for specific stream - status, exists := StreamStatusStore.Get(streamID) - if !exists { - respondWithError(w, "Stream status not found", http.StatusNotFound) - return - } - - jsonData, err := json.Marshal(status) - if err != nil { - respondWithError(w, "Failed to marshal status", http.StatusInternalServerError) - return - } - respondJsonOk(w, jsonData) - } -} diff --git a/server/ai_mediaserver.go b/server/ai_mediaserver.go index c3042a130..5e0ccc2af 100644 --- a/server/ai_mediaserver.go +++ b/server/ai_mediaserver.go @@ -87,6 +87,9 @@ func startAIMediaServer(ls *LivepeerServer) error { ls.HTTPMux.Handle("/live/video-to-video/{prefix}/{stream}/start", ls.StartLiveVideo()) ls.HTTPMux.Handle("/live/video-to-video/{stream}/update", ls.UpdateLiveVideo()) + // Stream status + ls.HTTPMux.Handle("/live/video-to-video/{streamId}/status", ls.GetLiveVideoToVideoStatus()) + return nil } @@ -581,6 +584,33 @@ func (ls *LivepeerServer) UpdateLiveVideo() http.Handler { }) } +func (ls *LivepeerServer) GetLiveVideoToVideoStatus() http.Handler { + return http.HandlerFunc(func(w http.ResponseWriter, r *http.Request) { + streamId := r.PathValue("streamId") + if streamId == "" { + http.Error(w, "stream id is required", http.StatusBadRequest) + return + } + + ctx := r.Context() + ctx = clog.AddVal(ctx, "stream", streamId) + + // Get status for specific stream + status, exists := StreamStatusStore.Get(streamId) + if !exists { + http.Error(w, "Stream not found", http.StatusNotFound) + return + } + + w.Header().Set("Content-Type", "application/json") + if err := json.NewEncoder(w).Encode(status); err != nil { + clog.Errorf(ctx, "Failed to encode stream status err=%v", err) + http.Error(w, "Failed to encode status", http.StatusInternalServerError) + return + } + }) +} + func (ls *LivepeerServer) cleanupLive(stream string) { ls.LivepeerNode.LiveMu.Lock() pub, ok := ls.LivepeerNode.LivePipelines[stream]