package server import ( "encoding/json" "fmt" "net/http" "time" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "sync-live/gen" "sync-live/internal/room" ) func (s *Server) writeProto(w http.ResponseWriter, m proto.Message, err error) { if err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } w.Header().Set("Content-Type", "application/json") data, merr := protojson.Marshal(m) if merr != nil { http.Error(w, merr.Error(), http.StatusInternalServerError) return } w.Write(data) } func (s *Server) handleConfig(w http.ResponseWriter, r *http.Request) { resp, err := s.svc.GetConfig(r.Context(), &gen.GetConfigRequest{}) s.writeProto(w, resp, err) } func (s *Server) handleRooms(w http.ResponseWriter, r *http.Request) { resp, err := s.svc.ListRooms(r.Context(), &gen.ListRoomsRequest{}) s.writeProto(w, resp, err) } func (s *Server) handlePublish(w http.ResponseWriter, r *http.Request) { var req gen.PublishRequest if err := protojson.Unmarshal(readBody(r), &req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } resp, err := s.svc.Publish(r.Context(), &req) s.writeProto(w, resp, err) } func (s *Server) handleSubscribe(w http.ResponseWriter, r *http.Request) { var req gen.SubscribeRequest if err := protojson.Unmarshal(readBody(r), &req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } resp, err := s.svc.Subscribe(r.Context(), &req) s.writeProto(w, resp, err) } func (s *Server) handleStop(w http.ResponseWriter, r *http.Request) { var req gen.StopStreamRequest if err := protojson.Unmarshal(readBody(r), &req); err != nil { http.Error(w, err.Error(), http.StatusBadRequest) return } resp, err := s.svc.StopStream(r.Context(), &req) s.writeProto(w, resp, err) } // handleRoomEvents 以 SSE 推送房间分发拓扑(分流状态),浏览器据此自动拉流。 func (s *Server) handleRoomEvents(w http.ResponseWriter, r *http.Request) { room := r.PathValue("room") flusher, ok := w.(http.Flusher) if !ok { http.Error(w, "streaming unsupported", http.StatusInternalServerError) return } w.Header().Set("Content-Type", "text/event-stream") w.Header().Set("Cache-Control", "no-cache") w.Header().Set("Connection", "keep-alive") ch, unsub := s.hub.subscribe(room) defer unsub() sendRoom := func(targets []*gen.StreamTarget) { data, _ := protojson.Marshal(&gen.RoomEvent{Room: room, Targets: targets}) fmt.Fprintf(w, "event: room\ndata: %s\n\n", data) flusher.Flush() } sendRoom(s.hub.targets(room)) ticker := time.NewTicker(20 * time.Second) defer ticker.Stop() for { select { case <-r.Context().Done(): return case msg := <-ch: sendRoom(decodeTargets(msg)) case <-ticker.C: sendRoom(s.hub.targets(room)) } } } func (s *Server) handleSRSStreams(w http.ResponseWriter, r *http.Request) { resp, err := http.Get(s.cfg.SRSBaseURL + "/api/v1/streams/") if err != nil { http.Error(w, "srs unreachable: "+err.Error(), http.StatusBadGateway) return } defer resp.Body.Close() var parsed struct { Code int `json:"code"` Streams []struct { App string `json:"app"` Name string `json:"name"` } `json:"streams"` } if err := json.NewDecoder(resp.Body).Decode(&parsed); err != nil { http.Error(w, "srs decode: "+err.Error(), http.StatusBadGateway) return } names := make([]string, 0, len(parsed.Streams)) for _, st := range parsed.Streams { names = append(names, st.Name) } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{"code": parsed.Code, "streams": names}) } func readBody(r *http.Request) []byte { buf := make([]byte, 0, 4096) tmp := make([]byte, 4096) for { n, err := r.Body.Read(tmp) if n > 0 { buf = append(buf, tmp[:n]...) } if err != nil { break } if len(buf) > 1<<20 { break } } return buf } func decodeTargets(msg []byte) []*gen.StreamTarget { var m struct { Targets map[string]*gen.StreamTarget `json:"targets"` } if err := json.Unmarshal(msg, &m); err != nil { return nil } out := make([]*gen.StreamTarget, 0, len(m.Targets)) for _, t := range m.Targets { out = append(out, t) } return out } func (s *Server) handleRoomsQuery(w http.ResponseWriter, r *http.Request) { q := r.URL.Query() page := 1 pageSize := 20 if v := q.Get("page"); v != "" { fmt.Sscanf(v, "%d", &page) } if v := q.Get("page_size"); v != "" { fmt.Sscanf(v, "%d", &pageSize) } if v := q.Get("pageSize"); v != "" { fmt.Sscanf(v, "%d", &pageSize) } search := q.Get("search") var searchPtr *string if search != "" { searchPtr = &search } statusStr := q.Get("status") var statusPtr *room.RoomStatus if statusStr != "" { if st, err := room.ParseRoomStatus(statusStr); err == nil { statusPtr = &st } } creator := q.Get("creator") var creatorPtr *string if creator != "" { creatorPtr = &creator } var isPublicPtr *bool if v := q.Get("is_public"); v != "" { b := v == "true" || v == "1" isPublicPtr = &b } sortBy := room.RoomListSortBy(q.Get("sort_by")) if sortBy == "" { sortBy = room.SortByCreatedAt } sortDir := room.SortDirection(q.Get("sort_direction")) if sortDir == "" { sortDir = room.SortDesc } query := room.RoomListQuery{ Pagination: room.PageParams{Page: page, PageSize: pageSize}, Search: searchPtr, Status: statusPtr, Creator: creatorPtr, IsPublic: isPublicPtr, SortBy: sortBy, SortDirection: sortDir, } // prefer DB-backed query if available if s.roomStore != nil && s.roomStore.DB() != nil { rooms, total, err := s.roomStore.QueryRooms(r.Context(), query) if err != nil { http.Error(w, err.Error(), http.StatusInternalServerError) return } w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ "rooms": rooms, "total": total, "page": page, "page_size": pageSize, }) return } // fallback to memory hub all := s.hub.list() w.Header().Set("Content-Type", "application/json") json.NewEncoder(w).Encode(map[string]interface{}{ "rooms": all, "total": len(all), "page": 1, "page_size": len(all), }) }