package server import ( "encoding/json" "fmt" "net/http" "time" "google.golang.org/protobuf/encoding/protojson" "google.golang.org/protobuf/proto" "gospeak-live-sfu-demo/gen" ) 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 }