From 2fca6e5c43fb8f759635f63ff4d9b6fc0b633414 Mon Sep 17 00:00:00 2001 From: liqi Date: Mon, 15 Dec 2025 13:21:08 +0800 Subject: [PATCH] =?UTF-8?q?=E5=BE=AE=E4=BF=A1=E5=B0=8F=E7=A8=8B=E5=BA=8F\?= =?UTF-8?q?=E6=8E=A8=E6=8B=89=E6=B5=81?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- internal/api/call_handler.go | 25 +++++- internal/mediaserver/room.go | 85 +++++++++---------- internal/middleware/request_log.go | 129 ++++++++++++++++++++++++++--- 3 files changed, 180 insertions(+), 59 deletions(-) diff --git a/internal/api/call_handler.go b/internal/api/call_handler.go index 73ef303..0afa302 100644 --- a/internal/api/call_handler.go +++ b/internal/api/call_handler.go @@ -59,9 +59,10 @@ type WebRTCICERequest struct { type JoinCallRoomResponse struct { RoomID string `json:"room_id"` Platform string `json:"platform"` - ICEServers []ICEServerConfig `json:"ice_servers,omitempty"` // H5/App 用 - PushURL string `json:"push_url,omitempty"` // 小程序用 - PullURLs []PullURLInfo `json:"pull_urls,omitempty"` // 小程序用 + ICEServers []ICEServerConfig `json:"ice_servers,omitempty"` // H5/App 用 + FlvPullURLs []PullURLInfo `json:"flv_pull_urls,omitempty"` // H5/App 拉取小程序流的 FLV 地址 + PushURL string `json:"push_url,omitempty"` // 小程序用 + PullURLs []PullURLInfo `json:"pull_urls,omitempty"` // 小程序用 Participants []ParticipantInfo `json:"participants"` } @@ -164,6 +165,20 @@ func JoinCallRoomHandler(c *gin.Context) { // 返回 ICE 服务器配置 response.ICEServers = getICEServers(req.UserID) + // 获取房间内小程序用户的 FLV 拉流地址（用于 Web 端播放小程序流） + flvPullURLs := make([]PullURLInfo, 0) + for _, p := range room.GetOtherParticipants(req.UserID) { + // 只返回小程序用户的 FLV 地址 + if p.Type == mediaserver.ParticipantTypeRTMP && p.FLVURL != "" { + flvPullURLs = append(flvPullURLs, PullURLInfo{ + UserID: p.UserID, + URL: p.PullURL, // RTMP 地址 + FLVURL: p.FLVURL, // HTTP-FLV 地址 + }) + } + } + response.FlvPullURLs = flvPullURLs + case "miniprogram": // 小程序使用 RTMP pType := mediaserver.ParticipantTypeRTMP @@ -187,9 +202,10 @@ func JoinCallRoomHandler(c *gin.Context) { } // 更新参与者的流地址 - room.SetParticipantRTMPURLs(req.UserID, stream.PushURL, stream.PullURL, stream.ID) + room.SetParticipantRTMPURLs(req.UserID, stream.PushURL, stream.PullURL, stream.FLVURL, stream.ID) participant.PushURL = stream.PushURL participant.PullURL = stream.PullURL + participant.FLVURL = stream.FLVURL participant.StreamID = stream.ID response.PushURL = stream.PushURL @@ -201,6 +217,7 @@ func JoinCallRoomHandler(c *gin.Context) { pullURLs = append(pullURLs, PullURLInfo{ UserID: p.UserID, URL: p.PullURL, + FLVURL: p.FLVURL, }) } } diff --git a/internal/mediaserver/room.go b/internal/mediaserver/room.go index 2ae7831..ca8cd5f 100644 --- a/internal/mediaserver/room.go +++ b/internal/mediaserver/room.go @@ -25,18 +25,19 @@ const ( // Participant 房间参与者 type Participant struct { - UserID string `json:"user_id"` - Type ParticipantType `json:"type"` - JoinedAt time.Time `json:"joined_at"` - HasAudio bool `json:"has_audio"` - HasVideo bool `json:"has_video"` - + UserID string `json:"user_id"` + Type ParticipantType `json:"type"` + JoinedAt time.Time `json:"joined_at"` + HasAudio bool `json:"has_audio"` + HasVideo bool `json:"has_video"` + // WebRTC 相关 PeerConnection interface{} `json:"-"` // *webrtc.PeerConnection - + // RTMP 相关 PushURL string `json:"push_url,omitempty"` // 推流地址 - PullURL string `json:"pull_url,omitempty"` // 拉流地址 + PullURL string `json:"pull_url,omitempty"` // 拉流地址 (RTMP) + FLVURL string `json:"flv_url,omitempty"` // HTTP-FLV 拉流地址 (给 Web 端用) StreamID string `json:"stream_id,omitempty"` // 流ID } @@ -46,24 +47,24 @@ type Room struct { CreatedAt time.Time `json:"created_at"` MaxSize int `json:"max_size"` Participants map[string]*Participant `json:"participants"` - + // 房间状态 - CallType string `json:"call_type"` // audio/video - IsGroupCall bool `json:"is_group_call"` - - mu sync.RWMutex - closeChan chan struct{} - closed bool + CallType string `json:"call_type"` // audio/video + IsGroupCall bool `json:"is_group_call"` + + mu sync.RWMutex + closeChan chan struct{} + closed bool } // RoomInfo 房间信息（用于API返回） type RoomInfo struct { - ID string `json:"id"` - CreatedAt time.Time `json:"created_at"` - ParticipantCount int `json:"participant_count"` - CallType string `json:"call_type"` - IsGroupCall bool `json:"is_group_call"` - Participants []ParticipantInfo `json:"participants"` + ID string `json:"id"` + CreatedAt time.Time `json:"created_at"` + ParticipantCount int `json:"participant_count"` + CallType string `json:"call_type"` + IsGroupCall bool `json:"is_group_call"` + Participants []ParticipantInfo `json:"participants"` } // ParticipantInfo 参与者信息（用于API返回） @@ -99,30 +100,30 @@ func (r *Room) SetCallType(callType string, isGroup bool) { func (r *Room) AddParticipant(userID string, pType ParticipantType) (*Participant, error) { r.mu.Lock() defer r.mu.Unlock() - + if r.closed { return nil, ErrRoomClosed } - + // 检查是否已存在 if p, exists := r.Participants[userID]; exists { return p, nil } - + // 检查房间容量 if len(r.Participants) >= r.MaxSize { return nil, ErrRoomFull } - + participant := &Participant{ UserID: userID, Type: pType, JoinedAt: time.Now(), } - + r.Participants[userID] = participant log.Printf("🎥 [Room:%s] 用户 %s 加入 (类型: %d)", r.ID, userID, pType) - + return participant, nil } @@ -130,7 +131,7 @@ func (r *Room) AddParticipant(userID string, pType ParticipantType) (*Participan func (r *Room) RemoveParticipant(userID string) { r.mu.Lock() defer r.mu.Unlock() - + if p, exists := r.Participants[userID]; exists { // 清理 PeerConnection if p.PeerConnection != nil { @@ -139,7 +140,7 @@ func (r *Room) RemoveParticipant(userID string) { delete(r.Participants, userID) log.Printf("🎥 [Room:%s] 用户 %s 离开", r.ID, userID) } - + // 如果房间空了，标记为可清理 if len(r.Participants) == 0 { log.Printf("🎥 [Room:%s] 房间已空，等待清理", r.ID) @@ -157,7 +158,7 @@ func (r *Room) GetParticipant(userID string) *Participant { func (r *Room) GetAllParticipants() []*Participant { r.mu.RLock() defer r.mu.RUnlock() - + participants := make([]*Participant, 0, len(r.Participants)) for _, p := range r.Participants { participants = append(participants, p) @@ -169,7 +170,7 @@ func (r *Room) GetAllParticipants() []*Participant { func (r *Room) GetOtherParticipants(excludeUserID string) []*Participant { r.mu.RLock() defer r.mu.RUnlock() - + participants := make([]*Participant, 0, len(r.Participants)-1) for _, p := range r.Participants { if p.UserID != excludeUserID { @@ -204,7 +205,7 @@ func (r *Room) IsFull() bool { func (r *Room) GetInfo() *RoomInfo { r.mu.RLock() defer r.mu.RUnlock() - + participants := make([]ParticipantInfo, 0, len(r.Participants)) for _, p := range r.Participants { typeName := "WebRTC" @@ -220,7 +221,7 @@ func (r *Room) GetInfo() *RoomInfo { JoinedAt: p.JoinedAt, }) } - + return &RoomInfo{ ID: r.ID, CreatedAt: r.CreatedAt, @@ -235,7 +236,7 @@ func (r *Room) GetInfo() *RoomInfo { func (r *Room) UpdateParticipantMedia(userID string, hasAudio, hasVideo bool) { r.mu.Lock() defer r.mu.Unlock() - + if p, exists := r.Participants[userID]; exists { p.HasAudio = hasAudio p.HasVideo = hasVideo @@ -243,13 +244,14 @@ func (r *Room) UpdateParticipantMedia(userID string, hasAudio, hasVideo bool) { } // SetParticipantRTMPURLs 设置参与者的RTMP推拉流地址 -func (r *Room) SetParticipantRTMPURLs(userID, pushURL, pullURL, streamID string) { +func (r *Room) SetParticipantRTMPURLs(userID, pushURL, pullURL, flvURL, streamID string) { r.mu.Lock() defer r.mu.Unlock() - + if p, exists := r.Participants[userID]; exists { p.PushURL = pushURL p.PullURL = pullURL + p.FLVURL = flvURL p.StreamID = streamID } } @@ -258,14 +260,14 @@ func (r *Room) SetParticipantRTMPURLs(userID, pushURL, pullURL, streamID string) func (r *Room) Close() { r.mu.Lock() defer r.mu.Unlock() - + if r.closed { return } - + r.closed = true close(r.closeChan) - + // 清理所有参与者 for userID, p := range r.Participants { if p.PeerConnection != nil { @@ -273,7 +275,7 @@ func (r *Room) Close() { } delete(r.Participants, userID) } - + log.Printf("🎥 [Room:%s] 已关闭", r.ID) } @@ -288,7 +290,7 @@ func (r *Room) IsClosed() bool { func (r *Room) Broadcast(excludeUserID string, message interface{}) { r.mu.RLock() defer r.mu.RUnlock() - + for userID := range r.Participants { if userID != excludeUserID { // TODO: 通过 WebSocket 发送消息 @@ -296,4 +298,3 @@ func (r *Room) Broadcast(excludeUserID string, message interface{}) { } } } - diff --git a/internal/middleware/request_log.go b/internal/middleware/request_log.go index fb49f03..7052c35 100644 --- a/internal/middleware/request_log.go +++ b/internal/middleware/request_log.go @@ -8,6 +8,7 @@ package middleware import ( "bytes" "encoding/json" + "fmt" "io" "strings" "time" @@ -18,6 +19,90 @@ import ( "gorm.io/gorm" ) +// 二进制 Content-Type 前缀列表 +var binaryContentTypes = []string{ + "image/", + "audio/", + "video/", + "application/octet-stream", + "application/pdf", + "application/zip", + "application/x-rar", + "application/x-7z", + "application/gzip", + "application/x-tar", + "font/", +} + +// 不需要记录响应体的路由前缀 +var skipResponseBodyRoutes = []string{ + "/uploads/", + "/static/", +} + +// isBinaryContentType 检测是否为二进制 Content-Type +func isBinaryContentType(contentType string) bool { + contentType = strings.ToLower(contentType) + for _, prefix := range binaryContentTypes { + if strings.HasPrefix(contentType, prefix) { + return true + } + } + return false +} + +// isBinaryData 检测数据是否为二进制（通过检查是否包含非 UTF-8 字符） +func isBinaryData(data []byte) bool { + if len(data) == 0 { + return false + } + // 检查前 512 字节是否包含二进制特征 + checkLen := len(data) + if checkLen > 512 { + checkLen = 512 + } + for i := 0; i < checkLen; i++ { + // 检测常见的二进制文件头 + if data[i] == 0 { + return true + } + } + // 检查是否以常见的二进制文件头开始 + if len(data) >= 2 { + // JPEG: FF D8 + if data[0] == 0xFF && data[1] == 0xD8 { + return true + } + // PNG: 89 50 + if data[0] == 0x89 && data[1] == 0x50 { + return true + } + // GIF: 47 49 + if data[0] == 0x47 && data[1] == 0x49 { + return true + } + // PDF: 25 50 + if data[0] == 0x25 && data[1] == 0x50 { + return true + } + // ZIP/DOCX/XLSX: 50 4B + if data[0] == 0x50 && data[1] == 0x4B { + return true + } + } + return false +} + +// shouldSkipResponseBody 检测是否应该跳过响应体记录 +func shouldSkipResponseBody(path string) bool { + for _, prefix := range skipResponseBodyRoutes { + if strings.HasPrefix(path, prefix) { + return true + } + } + return false +} + // RequestLogMiddleware 请求日志中间件 type RequestLogMiddleware struct { DB *gorm.DB @@ -39,7 +124,7 @@ func (m *RequestLogMiddleware) Handler() gin.HandlerFunc { // 获取请求IP ip := getClientIP(c) - + // 本地IP不记录 if utils.GetIPLocation(ip) == "本地" { c.Next() @@ -67,7 +152,7 @@ func (m *RequestLogMiddleware) Handler() gin.HandlerFunc { // 创建响应写入器 writer := &responseWriter{ ResponseWriter: c.Writer, - body: &bytes.Buffer{}, + body: &bytes.Buffer{}, } c.Writer = writer @@ -89,24 +174,43 @@ func (m *RequestLogMiddleware) logRequest(c *gin.Context, ip, userID string, req // 获取响应code（从响应体中解析） responseCode := 0 - if len(responseBody) > 0 { - // 尝试解析响应体获取code + if len(responseBody) > 0 && !isBinaryData(responseBody) { + // 尝试解析响应体获取code（仅对非二进制数据） var resp model.ApiResponse if err := json.Unmarshal(responseBody, &resp); err == nil { responseCode = resp.Code } } - // 限制请求参数长度（避免存储过大） - requestParams := string(requestBody) - if len(requestParams) > 5000 { - requestParams = requestParams[:5000] + "...(truncated)" + // 处理请求参数 + var requestParams string + requestContentType := c.GetHeader("Content-Type") + if isBinaryContentType(requestContentType) || isBinaryData(requestBody) { + // 二进制请求体，只记录大小 + requestParams = fmt.Sprintf("[binary data: %d bytes]", len(requestBody)) + } else { + requestParams = string(requestBody) + if len(requestParams) > 5000 { + requestParams = requestParams[:5000] + "...(truncated)" + } } - // 限制返回参数长度 - responseParams := string(responseBody) - if len(responseParams) > 5000 { - responseParams = responseParams[:5000] + "...(truncated)" + // 处理响应参数 + var responseParams string + responseContentType := c.Writer.Header().Get("Content-Type") + requestPath := c.Request.URL.Path + + if shouldSkipResponseBody(requestPath) { + // 静态文件路由，跳过响应体记录 + responseParams = fmt.Sprintf("[static file: %d bytes]", len(responseBody)) + } else if isBinaryContentType(responseContentType) || isBinaryData(responseBody) { + // 二进制响应体，只记录大小 + responseParams = fmt.Sprintf("[binary data: %d bytes]", len(responseBody)) + } else { + responseParams = string(responseBody) + if len(responseParams) > 5000 { + responseParams = responseParams[:5000] + "...(truncated)" + } } // 创建日志记录 @@ -170,4 +274,3 @@ func (w *responseWriter) WriteHeader(statusCode int) { w.status = statusCode w.ResponseWriter.WriteHeader(statusCode) } -