视频会议保存
This commit is contained in:
duoaohui
@@ -69,6 +69,39 @@ func memberStateKey(roomCode string, userID int64) string {
|
||||
return fmt.Sprintf("echo:meeting:member_state:%s:%d", roomCode, userID)
|
||||
}
|
||||
|
||||
// screenOwnerKey Redis String，记录当前会议正在共享屏幕的人 + Producer ID
|
||||
// 值格式："{userId}:{producerId}"；为空 / Nil 表示当前无人共享
|
||||
// 单字符串而非 Hash 是因为"会议同时只允许 1 份屏幕共享"，无需多字段
|
||||
//
|
||||
// 生命周期：
|
||||
// - OnProduceStart 收到 appData.screen=true 时 SET（同时广播 meeting.screen.started）
|
||||
// - OnProducerClose / cleanupUserResources 命中该 Producer 时 DEL（同时广播 meeting.screen.stopped）
|
||||
// - 房间销毁路径（cleanupRoomRedisResidual）一并 Del
|
||||
func screenOwnerKey(roomCode string) string {
|
||||
return fmt.Sprintf("echo:meeting:screen_owner:%s", roomCode)
|
||||
}
|
||||
|
||||
// formatScreenOwnerValue / parseScreenOwnerValue 统一 screenOwnerKey 的取值格式
|
||||
// 复用于 set / parse 双向，避免格式漂移
|
||||
func formatScreenOwnerValue(userID int64, producerID string) string {
|
||||
return fmt.Sprintf("%d:%s", userID, producerID)
|
||||
}
|
||||
|
||||
func parseScreenOwnerValue(raw string) (int64, string, bool) {
|
||||
if raw == "" {
|
||||
return 0, "", false
|
||||
}
|
||||
parts := strings.SplitN(raw, ":", 2)
|
||||
if len(parts) != 2 || parts[1] == "" {
|
||||
return 0, "", false
|
||||
}
|
||||
var uid int64
|
||||
if _, err := fmt.Sscanf(parts[0], "%d", &uid); err != nil || uid <= 0 {
|
||||
return 0, "", false
|
||||
}
|
||||
return uid, parts[1], true
|
||||
}
|
||||
|
||||
// resourceTTL 单个用户资源追踪集合 TTL
|
||||
// 设计：会议期间维持可达即可；若用户长期不活跃由断线清理接管
|
||||
// Task 16 Nit：常量已迁出至 constants.MeetingResourceTrackTTLSeconds，此处保留计算式 wrapper 方便调用侧零改动
|
||||
@@ -237,6 +270,43 @@ func (s *MeetingSignalService) OnRoomJoin(ctx context.Context, userID int64, roo
|
||||
func (s *MeetingSignalService) pushExistingRoomState(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||||
s.pushExistingProducers(ctx, roomID, roomCode, userID)
|
||||
s.pushExistingMemberStates(ctx, roomID, roomCode, userID)
|
||||
s.pushExistingScreenShare(ctx, roomCode, userID)
|
||||
}
|
||||
|
||||
// pushExistingScreenShare 后入者定向补推当前屏幕共享状态（Phase 3）
|
||||
// 复用 meeting.screen.started 事件语义，前端共用同一 handler，无需新事件类型
|
||||
// 没有人在共享 / Redis 异常 / 解析失败 都静默 no-op
|
||||
func (s *MeetingSignalService) pushExistingScreenShare(ctx context.Context, roomCode string, userID int64) {
|
||||
funcName := "service.meeting_signal_service.pushExistingScreenShare"
|
||||
|
||||
raw, err := s.redis.Get(ctx, screenOwnerKey(roomCode)).Result()
|
||||
if err != nil {
|
||||
if err != redis.Nil {
|
||||
logs.Warn(ctx, funcName, "读取屏幕共享拥有者失败（忽略）",
|
||||
zap.String("room_code", roomCode), zap.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
ownerUID, producerID, ok := parseScreenOwnerValue(raw)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
if ownerUID == userID {
|
||||
// 自己就是共享者（理论上不会发生：刚 join 不应已是 owner），跳过避免回响
|
||||
return
|
||||
}
|
||||
if err := s.broadcaster.PublishToUser(ctx, userID, constants.MeetingWSEventScreenStarted, map[string]interface{}{
|
||||
"room_code": roomCode,
|
||||
"user_id": ownerUID,
|
||||
"producer_id": producerID,
|
||||
"existing": true,
|
||||
}); err != nil {
|
||||
logs.Warn(ctx, funcName, "定向推送 existing screen.started 失败",
|
||||
zap.Int64("to_user", userID),
|
||||
zap.Int64("owner_user", ownerUID),
|
||||
zap.String("producer_id", producerID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
// pushExistingProducers 向刚加入者定向推送房间里其他用户已产生的 producer 列表
|
||||
@@ -384,7 +454,7 @@ func (s *MeetingSignalService) OnWSDisconnect(ctx context.Context, userID int64)
|
||||
return
|
||||
}
|
||||
|
||||
s.cleanupUserResources(ctx, room.RoomCode, userID)
|
||||
s.cleanupUserResources(ctx, room.ID, room.RoomCode, userID)
|
||||
|
||||
if s.lifecycleSvc != nil && room.HostID == userID {
|
||||
s.lifecycleSvc.OnHostDisconnect(ctx, room.RoomCode, userID)
|
||||
@@ -405,7 +475,7 @@ func (s *MeetingSignalService) OnRoomLeave(ctx context.Context, userID int64, ro
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
s.cleanupUserResources(ctx, roomCode, userID)
|
||||
s.cleanupUserResources(ctx, room.ID, roomCode, userID)
|
||||
|
||||
// P2-8 修复：使用常量 MeetingLeftReasonDisconnect，避免 "ws_disconnect" 等硬编码
|
||||
// 与前端 MEETING_LEFT_REASON_LABEL 字面值不一致
|
||||
@@ -524,6 +594,36 @@ type ProduceStartPayload struct {
|
||||
TransportID string `json:"transport_id"`
|
||||
Kind string `json:"kind"` // "audio" | "video"
|
||||
RtpParameters json.RawMessage `json:"rtp_parameters"`
|
||||
// AppData 客户端透传给 mediasoup Producer 的自定义元信息
|
||||
// Phase 3 屏幕共享引入：约定 {"screen": true} 表示该 video Producer 是屏幕分享流；
|
||||
// 服务端识别后会写入 screenOwnerKey + 广播 meeting.screen.started，前端将该流提升大画面。
|
||||
// 服务端会强制把 userId / roomCode 注入回 appData，覆盖客户端伪造尝试（见 HTTPMediaOrchestrator.CreateProducer）。
|
||||
AppData json.RawMessage `json:"app_data,omitempty"`
|
||||
}
|
||||
|
||||
// isScreenAppData 判断客户端 app_data 是否声明了屏幕共享标识
|
||||
// 容错：raw 为空 / 非对象 / 字段缺失均返回 false
|
||||
func isScreenAppData(raw json.RawMessage) bool {
|
||||
if len(raw) == 0 {
|
||||
return false
|
||||
}
|
||||
var m map[string]any
|
||||
if err := json.Unmarshal(raw, &m); err != nil {
|
||||
return false
|
||||
}
|
||||
v, ok := m["screen"]
|
||||
if !ok {
|
||||
return false
|
||||
}
|
||||
switch x := v.(type) {
|
||||
case bool:
|
||||
return x
|
||||
case string:
|
||||
return x == "true" || x == "1"
|
||||
case float64:
|
||||
return x != 0
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// ProduceStartResult 返回给客户端的 producerID
|
||||
@@ -544,25 +644,61 @@ func (s *MeetingSignalService) OnProduceStart(ctx context.Context, userID int64,
|
||||
if err := s.assertOwnsResource(ctx, payload.RoomCode, userID, "transport", payload.TransportID); err != nil {
|
||||
return nil, err
|
||||
}
|
||||
|
||||
// Phase 3：识别屏幕共享意图，强制 video kind 才允许（音频流不参与屏幕共享语义）
|
||||
isScreen := payload.Kind == "video" && isScreenAppData(payload.AppData)
|
||||
|
||||
// Phase 3：单会议同时仅允许 1 份屏幕共享。若已有他人在共享，直接拒绝；
|
||||
// 同一用户重复请求（例如客户端重发）则容忍，由后续 SET 覆盖
|
||||
if isScreen {
|
||||
if existingRaw, err := s.redis.Get(ctx, screenOwnerKey(payload.RoomCode)).Result(); err == nil {
|
||||
if ownerUID, _, ok := parseScreenOwnerValue(existingRaw); ok && ownerUID != userID {
|
||||
return nil, fmt.Errorf("当前会议已有成员正在共享屏幕")
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
producerID, err := s.mediaOrchestrator.CreateProducer(ctx, &CreateProducerReq{
|
||||
RoomCode: payload.RoomCode,
|
||||
UserID: userID,
|
||||
TransportID: payload.TransportID,
|
||||
Kind: payload.Kind,
|
||||
RtpParameters: payload.RtpParameters,
|
||||
AppData: payload.AppData,
|
||||
})
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.trackResource(ctx, payload.RoomCode, userID, "producer", producerID)
|
||||
|
||||
// Phase 3：屏幕共享额外维护 owner key + 触发 meeting.screen.started 广播
|
||||
// 注意：先广播 producer.new（保证消费者侧 Consumer 创建），再广播 screen.started（标记大画面提升）
|
||||
if isScreen {
|
||||
if err := s.redis.Set(ctx, screenOwnerKey(payload.RoomCode), formatScreenOwnerValue(userID, producerID), resourceTTL).Err(); err != nil {
|
||||
logs.Warn(ctx, "service.meeting_signal_service.OnProduceStart", "写入屏幕共享拥有者失败（不阻断业务）",
|
||||
zap.String("room_code", payload.RoomCode),
|
||||
zap.Int64("user_id", userID),
|
||||
zap.String("producer_id", producerID),
|
||||
zap.Error(err))
|
||||
}
|
||||
}
|
||||
|
||||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberProducerNew, map[string]interface{}{
|
||||
"room_code": payload.RoomCode,
|
||||
"user_id": userID,
|
||||
"producer_id": producerID,
|
||||
"kind": payload.Kind,
|
||||
"screen": isScreen,
|
||||
}, userID)
|
||||
|
||||
if isScreen {
|
||||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventScreenStarted, map[string]interface{}{
|
||||
"room_code": payload.RoomCode,
|
||||
"user_id": userID,
|
||||
"producer_id": producerID,
|
||||
}) // 不排除任何人：共享者本人也接收，便于前端统一更新大画面 UI
|
||||
}
|
||||
|
||||
return &ProduceStartResult{ProducerID: producerID}, nil
|
||||
}
|
||||
|
||||
@@ -653,6 +789,9 @@ func (s *MeetingSignalService) OnProducerClose(ctx context.Context, userID int64
|
||||
}
|
||||
s.untrackResource(ctx, payload.RoomCode, userID, "producer", payload.ProducerID)
|
||||
|
||||
// Phase 3：若关闭的恰好是当前屏幕共享 Producer，则一并清 screenOwnerKey + 广播 screen.stopped
|
||||
s.maybeReleaseScreenOwner(ctx, room.ID, payload.RoomCode, userID, payload.ProducerID)
|
||||
|
||||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), room.ID, constants.MeetingWSEventMemberProducerNew, map[string]interface{}{
|
||||
"room_code": payload.RoomCode,
|
||||
"user_id": userID,
|
||||
@@ -662,11 +801,57 @@ func (s *MeetingSignalService) OnProducerClose(ctx context.Context, userID int64
|
||||
return nil
|
||||
}
|
||||
|
||||
// maybeReleaseScreenOwner 检查并释放屏幕共享拥有者状态
|
||||
// 当 producerID 恰好等于 screenOwnerKey 当前值的 producer 段时，认为这次关闭是屏幕分享停止：
|
||||
// - DEL screenOwnerKey
|
||||
// - 广播 meeting.screen.stopped 给所有活跃成员（含共享者本人）
|
||||
//
|
||||
// 非屏幕 Producer 或 key 已被他人覆盖时静默 no-op，保证幂等
|
||||
func (s *MeetingSignalService) maybeReleaseScreenOwner(ctx context.Context, roomID int64, roomCode string, userID int64, producerID string) {
|
||||
funcName := "service.meeting_signal_service.maybeReleaseScreenOwner"
|
||||
|
||||
key := screenOwnerKey(roomCode)
|
||||
raw, err := s.redis.Get(ctx, key).Result()
|
||||
if err != nil {
|
||||
// redis.Nil（无人共享）属正常路径，其他错误仅 Warn 不阻塞
|
||||
if err != redis.Nil {
|
||||
logs.Warn(ctx, funcName, "读取屏幕共享拥有者失败（忽略）",
|
||||
zap.String("room_code", roomCode), zap.Error(err))
|
||||
}
|
||||
return
|
||||
}
|
||||
ownerUID, ownerProducerID, ok := parseScreenOwnerValue(raw)
|
||||
if !ok || ownerProducerID != producerID {
|
||||
return
|
||||
}
|
||||
// 防御性日志：理论上 ownerUID 必然等于 userID（trackResource 已校验归属）
|
||||
if ownerUID != userID {
|
||||
logs.Warn(ctx, funcName, "屏幕共享拥有者与关闭者不一致（仍按停止处理）",
|
||||
zap.Int64("owner_uid", ownerUID),
|
||||
zap.Int64("closer_uid", userID),
|
||||
zap.String("producer_id", producerID))
|
||||
}
|
||||
|
||||
if err := s.redis.Del(ctx, key).Err(); err != nil {
|
||||
logs.Warn(ctx, funcName, "删除屏幕共享拥有者失败（忽略）",
|
||||
zap.String("room_code", roomCode), zap.Error(err))
|
||||
}
|
||||
|
||||
go s.broadcaster.BroadcastToMeeting(logs.DetachContext(ctx), roomID, constants.MeetingWSEventScreenStopped, map[string]interface{}{
|
||||
"room_code": roomCode,
|
||||
"user_id": ownerUID,
|
||||
"producer_id": producerID,
|
||||
})
|
||||
}
|
||||
|
||||
// ========== 资源清理 ==========
|
||||
|
||||
// cleanupUserResources 批量关闭指定用户在某会议的所有媒体资源
|
||||
// WS 断开、主动离会、被踢时使用；依赖 Redis 集合中追踪的资源 ID
|
||||
func (s *MeetingSignalService) cleanupUserResources(ctx context.Context, roomCode string, userID int64) {
|
||||
//
|
||||
// Phase 3 屏幕共享：调用方需提供 roomID，便于在该用户恰好是屏幕共享者时触发
|
||||
// meeting.screen.stopped 广播。roomID==0 时跳过屏幕广播（仅清 Redis），保持向后兼容。
|
||||
func (s *MeetingSignalService) cleanupUserResources(ctx context.Context, roomID int64, roomCode string, userID int64) {
|
||||
funcName := "service.meeting_signal_service.cleanupUserResources"
|
||||
|
||||
key := resourceTrackKey(roomCode, userID)
|
||||
@@ -687,6 +872,11 @@ func (s *MeetingSignalService) cleanupUserResources(ctx context.Context, roomCod
|
||||
switch kind {
|
||||
case "producer":
|
||||
_ = s.mediaOrchestrator.CloseProducer(ctx, id)
|
||||
// Phase 3：若该 Producer 是当前屏幕共享流，释放 owner key 并广播 screen.stopped
|
||||
// 复用 OnProducerClose 同款幂等逻辑；roomID==0 跳过避免无效广播
|
||||
if roomID > 0 {
|
||||
s.maybeReleaseScreenOwner(ctx, roomID, roomCode, userID, id)
|
||||
}
|
||||
case "consumer":
|
||||
_ = s.mediaOrchestrator.CloseConsumer(ctx, id)
|
||||
case "transport":
|
||||
@@ -714,7 +904,14 @@ func (s *MeetingSignalService) cleanupUserResources(ctx context.Context, roomCod
|
||||
// - package 私有：允许 meeting_service / meeting_lifecycle_service 等同包文件直接调用
|
||||
func cleanupRoomRedisResidual(ctx context.Context, rdb *redis.Client, roomCode string, userIDs []int64) {
|
||||
funcName := "service.meeting_signal_service.cleanupRoomRedisResidual"
|
||||
if rdb == nil || len(userIDs) == 0 {
|
||||
if rdb == nil {
|
||||
return
|
||||
}
|
||||
// Phase 3：会议销毁路径必删 screenOwnerKey（即使无活跃用户也清理）
|
||||
// 单独 Del 不依赖 pipeline，因为会议销毁是低频路径，多一次 RTT 可接受
|
||||
_ = rdb.Del(ctx, screenOwnerKey(roomCode)).Err()
|
||||
|
||||
if len(userIDs) == 0 {
|
||||
return
|
||||
}
|
||||
pipe := rdb.Pipeline()
|
||||
|
||||
Reference in New Issue
Block a user