package server import ( "context" "fmt" "time" "gospeak-live-sfu-demo/gen" "gospeak-live-sfu-demo/internal/config" "gospeak-live-sfu-demo/internal/sfu/cloudflare" "gospeak-live-sfu-demo/internal/sfu/srs" ) // Service 实现 LiveSFU 控制面逻辑:协调房间在各 SFU 后端的发布/订阅拓扑。 // 媒体面(SDP 交换)由浏览器经反向代理直连 Cloudflare / SRS,本服务只持有凭证。 type Service struct { cfg *config.Config cf *cloudflare.Provider srsP *srs.Provider hub *roomHub } func NewService(cfg *config.Config, cf *cloudflare.Provider, srsP *srs.Provider, hub *roomHub) *Service { return &Service{cfg: cfg, cf: cf, srsP: srsP, hub: hub} } func backendName(k gen.BackendKind) string { switch k { case gen.BackendKind_BACKEND_KIND_CLOUDFLARE: return "cloudflare" case gen.BackendKind_BACKEND_KIND_SRS: return "srs" default: return "" } } // GetConfig 返回后端能力与全局拓扑信息(浏览器据此渲染分发面板)。 func (s *Service) GetConfig(ctx context.Context, req *gen.GetConfigRequest) (*gen.GetConfigResponse, error) { resp := &gen.GetConfigResponse{ Candidate: s.srsP.Candidate(), TokenRequired: s.cfg.TokenRequired, } for _, name := range s.cfg.ProviderList() { switch name { case "cloudflare": resp.Backends = append(resp.Backends, s.cf.BackendInfo()) case "srs": resp.Backends = append(resp.Backends, s.srsP.BackendInfo()) } } return resp, nil } func (s *Service) ListRooms(ctx context.Context, req *gen.ListRoomsRequest) (*gen.ListRoomsResponse, error) { return &gen.ListRoomsResponse{Rooms: s.hub.list()}, nil } // Publish 开始向某后端发布:Cloudflare 创建 session;SRS 分配 stream + 签发 JWT。 func (s *Service) Publish(ctx context.Context, req *gen.PublishRequest) (*gen.PublishResponse, error) { room := req.GetRoom() if room == "" { return nil, fmt.Errorf("room required") } backend := backendName(req.GetBackend()) if backend == "" { return nil, fmt.Errorf("unknown backend") } identity := req.GetIdentity() if identity == "" { identity = randomID() } switch backend { case "cloudflare": if !s.cf.Configured() { return nil, fmt.Errorf("cloudflare realtime not configured (set CF_APP_ID / CF_APP_SECRET)") } sessionID, err := s.cf.Client().CreateSession(room) if err != nil { return nil, fmt.Errorf("cf create session: %w", err) } info, err := s.cf.Client().GetSession(sessionID) if err != nil { return nil, fmt.Errorf("cf get session: %w", err) } target := &gen.StreamTarget{ Backend: gen.BackendKind_BACKEND_KIND_CLOUDFLARE, SessionId: sessionID, PublishedAt: time.Now().Unix(), } s.hub.setTarget(room, "cloudflare", target) return &gen.PublishResponse{ SessionId: sessionID, IceServers: iceServersToProto(info.IceServers), Target: target, }, nil case "srs": stream := "live-" + room token, _ := signStreamToken(s.cfg.TokenSecret, stream, identity, "publish", 2*time.Hour) target := &gen.StreamTarget{ Backend: gen.BackendKind_BACKEND_KIND_SRS, Stream: stream, PublishToken: token, Url: fmt.Sprintf("/rtc/v1/whep/?app=%s&stream=%s", s.srsP.App(), stream), PublishedAt: time.Now().Unix(), } s.hub.setTarget(room, "srs", target) return &gen.PublishResponse{ Stream: stream, PublishToken: token, IceServers: []*gen.IceServer{{Urls: []string{s.cf.Stun()}}}, Target: target, }, nil } return nil, fmt.Errorf("unsupported backend") } // Subscribe 订阅某房间在某后端的分发目标,返回建立 WebRTC 所需的 session / stream。 func (s *Service) Subscribe(ctx context.Context, req *gen.SubscribeRequest) (*gen.SubscribeResponse, error) { room := req.GetRoom() if room == "" { return nil, fmt.Errorf("room required") } backend := backendName(req.GetBackend()) if backend == "" { return nil, fmt.Errorf("unknown backend") } pub := s.findTarget(room, req.GetBackend()) if pub == nil { return nil, fmt.Errorf("room %q not live on %s", room, backend) } switch backend { case "cloudflare": viewerSession, err := s.cf.Client().CreateSession(room) if err != nil { return nil, fmt.Errorf("cf create viewer session: %w", err) } info, err := s.cf.Client().GetSession(viewerSession) if err != nil { return nil, fmt.Errorf("cf get viewer session: %w", err) } return &gen.SubscribeResponse{ SessionId: viewerSession, PublisherSessionId: pub.GetSessionId(), IceServers: iceServersToProto(info.IceServers), }, nil case "srs": return &gen.SubscribeResponse{ Stream: pub.GetStream(), IceServers: []*gen.IceServer{{Urls: []string{s.cf.Stun()}}}, }, nil } return nil, fmt.Errorf("unsupported backend") } func (s *Service) StopStream(ctx context.Context, req *gen.StopStreamRequest) (*gen.StopStreamResponse, error) { room := req.GetRoom() backend := backendName(req.GetBackend()) if room == "" || backend == "" { return nil, fmt.Errorf("room and backend required") } if backend == "cloudflare" { for _, t := range s.hub.targets(room) { if t.GetBackend() == gen.BackendKind_BACKEND_KIND_CLOUDFLARE && t.GetSessionId() != "" { _ = s.cf.Client().DeleteSession(t.GetSessionId()) } } } s.hub.removeTarget(room, backend) return &gen.StopStreamResponse{Ok: true}, nil } // WatchRoom 服务端流式推送房间分发拓扑(分流)变化。 func (s *Service) WatchRoom(req *gen.WatchRoomRequest, stream gen.LiveSFU_WatchRoomServer) error { room := req.GetRoom() ch, unsub := s.hub.subscribe(room) defer unsub() if err := stream.Send(&gen.RoomEvent{Room: room, Targets: s.hub.targets(room)}); err != nil { return err } ticker := time.NewTicker(20 * time.Second) defer ticker.Stop() for { select { case <-stream.Context().Done(): return nil case msg := <-ch: if err := stream.Send(&gen.RoomEvent{Room: room, Targets: decodeTargets(msg)}); err != nil { return err } case <-ticker.C: if err := stream.Send(&gen.RoomEvent{Room: room, Targets: s.hub.targets(room)}); err != nil { return err } } } } func (s *Service) findTarget(room string, kind gen.BackendKind) *gen.StreamTarget { for _, t := range s.hub.targets(room) { if t.GetBackend() == kind { return t } } return nil } func iceServersToProto(in []cloudflare.IceServer) []*gen.IceServer { out := make([]*gen.IceServer, 0, len(in)) for _, s := range in { out = append(out, &gen.IceServer{Urls: s.URLs, Username: s.Username, Credential: s.Credential}) } return out }