live-sfu-demo/internal/server/gateway.go

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
}