yvan/kra-new
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

对接

Browse Source
This commit is contained in:
Yvan
2026-09-23 17:42:55 +08:00
parent 435b601273
commit d24eb2efd5
3 changed files with 58 additions and 3 deletions
Download Patch File Download Diff File Expand all files Collapse all files
internal/integration/live/server.go
+54 -3
View File
@@ -137,14 +137,23 @@ func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.Room
if err != nil {
return nil, nil, err
}
if logger == nil {
logger = slog.Default()
}
logger = logger.With("mod", "live")
m := &Manager{hub: hub, streams: streams, roomCatalog: rooms, chat: chat, logger: logger, api: api, rooms: make(map[string]*room), peers: make(map[*melody.Session]*peer), observers: make(map[*melody.Session]struct{}), stop: make(chan struct{})}
hub.OnMessage(m.onMessage)
hub.OnDisconnect(m.onDisconnect)
go m.heartbeatLoop()
logger.Info("live signaling manager initialized",
"stun_configured", len(liveICEServers()) > 0,
"udp_min_port", envPort("KRA_LIVE_UDP_MIN_PORT"),
"udp_max_port", envPort("KRA_LIVE_UDP_MAX_PORT"))
return m, m.Close, nil
}
func (m *Manager) Close() {
m.liveLogger().Info("live signaling manager stopping")
m.stopOnce.Do(func() { close(m.stop) })
m.mu.Lock()
peers := make([]*peer, 0, len(m.peers))
@@ -160,6 +169,7 @@ func (m *Manager) Close() {
func (m *Manager) onMessage(session *melody.Session, raw []byte) {
var msg message
if err := json.Unmarshal(raw, &msg); err != nil {
m.liveLogger().Warn("live signaling message rejected", "reason", "invalid_json", "message_bytes", len(raw), "remote_addr", sessionRemoteAddr(session))
m.send(session, "Error", map[string]any{"code": "INVALID_JSON", "message": "invalid json"})
return
}
@@ -167,6 +177,7 @@ func (m *Manager) onMessage(session *melody.Session, raw []byte) {
case "Join":
var payload joinPayload
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
m.liveLogger().Warn("live join rejected", "code", "INVALID_JOIN", "reason", "invalid_payload", "remote_addr", sessionRemoteAddr(session))
m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "invalid join payload"})
return
}
@@ -174,6 +185,7 @@ func (m *Manager) onMessage(session *melody.Session, raw []byte) {
case "Offer":
var payload sessionDescriptionPayload
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
m.liveLogger().Warn("live offer rejected", "code", "INVALID_OFFER", "reason", "invalid_payload", "remote_addr", sessionRemoteAddr(session))
m.send(session, "Error", map[string]any{"code": "INVALID_OFFER", "message": "invalid offer payload"})
return
}
@@ -203,6 +215,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
role := strings.ToLower(strings.TrimSpace(payload.Role))
streamID := strings.TrimSpace(payload.StreamID)
if role == "observer" {
m.liveLogger().Info("live observer joined", "remote_addr", sessionRemoteAddr(session))
m.mu.Lock()
m.observers[session] = struct{}{}
m.mu.Unlock()
@@ -211,11 +224,14 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
return
}
if streamID == "" || (role != "publisher" && role != "subscriber") {
m.liveLogger().Warn("live join rejected", "code", "INVALID_JOIN", "role", role, "stream_id", streamID, "remote_addr", sessionRemoteAddr(session))
m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "role and streamId are required"})
return
}
userID, authenticated := sessionUserID(session)
m.liveLogger().Info("live join requested", "role", role, "stream_id", streamID, "user_id", userID, "authenticated", authenticated, "remote_addr", sessionRemoteAddr(session))
if role == "publisher" && !authenticated {
m.liveLogger().Warn("live join rejected", "code", "AUTH_REQUIRED", "role", role, "stream_id", streamID, "remote_addr", sessionRemoteAddr(session))
m.send(session, "JoinRejected", map[string]any{"code": "AUTH_REQUIRED", "message": "authenticated user is required"})
return
}
@@ -227,10 +243,12 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
}
if role == "publisher" {
if err := m.authorizeRoomPublisher(streamID, userID); err != nil {
m.liveLogger().Warn("live join rejected", "code", "PUBLISH_FORBIDDEN", "role", role, "stream_id", streamID, "user_id", userID, "error", err)
m.send(session, "JoinRejected", map[string]any{"code": "PUBLISH_FORBIDDEN", "message": err.Error()})
return
}
} else if err := m.authorizeRoomSubscriber(streamID); err != nil {
m.liveLogger().Warn("live join rejected", "code", "ROOM_UNAVAILABLE", "role", role, "stream_id", streamID, "user_id", userID, "error", err)
m.send(session, "JoinRejected", map[string]any{"code": "ROOM_UNAVAILABLE", "message": err.Error()})
return
}
@@ -240,10 +258,12 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
occupied := r.publisher != nil
r.mu.RUnlock()
if occupied {
m.liveLogger().Warn("live join rejected", "code", "DUPLICATE_PUBLISHER", "role", role, "stream_id", streamID, "user_id", userID)
m.send(session, "JoinRejected", map[string]any{"code": "DUPLICATE_PUBLISHER", "message": "stream already has a publisher"})
return
}
if _, err := m.streams.RegisterPublisher(context.Background(), streamID, userID); err != nil {
m.liveLogger().Warn("live join rejected", "code", errorCode(err), "role", role, "stream_id", streamID, "user_id", userID, "error", err)
m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
return
}
@@ -259,6 +279,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
m.peers[session] = p
m.mu.Unlock()
m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "mode": "pion-sfu"})
m.liveLogger().Info("live publisher joined", "stream_id", streamID, "user_id", userID, "remote_addr", sessionRemoteAddr(session))
m.sendPublisherStatus(streamID, false, "publisher_waiting_for_offer")
m.broadcastStreamList()
return
@@ -280,6 +301,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
m.closePeer(old, "logged_in_elsewhere")
}
if _, err := m.streams.RegisterSubscriber(context.Background(), streamID, userID); err != nil {
m.liveLogger().Warn("live join rejected", "code", errorCode(err), "role", role, "stream_id", streamID, "user_id", userID, "error", err)
m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
return
}
@@ -297,6 +319,7 @@ func (m *Manager) join(session *melody.Session, payload joinPayload) {
publicUserID = 0
}
m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "subscriberId": p.subscriberID, "userId": publicUserID, "mode": "pion-sfu"})
m.liveLogger().Info("live subscriber joined", "stream_id", streamID, "user_id", publicUserID, "subscriber_id", p.subscriberID, "authenticated", authenticated, "remote_addr", sessionRemoteAddr(session))
m.sendPublisherStatusTo(session, streamID, ready, "publisher_offline")
if ready {
m.send(session, "PublisherStatus", map[string]any{"streamId": streamID, "ready": true})
@@ -414,15 +437,19 @@ func (m *Manager) authorizeRoomSubscriber(streamID string) error {
func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptionPayload) {
p := m.peer(session)
if p == nil || (p.role != "publisher" && p.role != "subscriber") {
m.liveLogger().Warn("live offer rejected", "code", "NOT_JOINED", "remote_addr", sessionRemoteAddr(session))
m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a stream first"})
return
}
if strings.TrimSpace(payload.SDP) == "" {
m.liveLogger().Warn("live offer rejected", "code", "INVALID_SDP", "role", p.role, "stream_id", p.streamID, "user_id", p.userID)
m.send(session, "Error", map[string]any{"code": "INVALID_SDP", "message": "sdp is required"})
return
}
m.liveLogger().Info("live offer received", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "sdp_bytes", len(payload.SDP))
pc, err := m.api.NewPeerConnection(webrtc.Configuration{ICEServers: liveICEServers()})
if err != nil {
m.liveLogger().Error("live peer connection creation failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err)
m.send(session, "Error", map[string]any{"code": "PEER_CONNECTION", "message": err.Error()})
return
}
@@ -449,31 +476,37 @@ func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptio
}
}
if err = pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: payload.SDP}); err != nil {
m.liveLogger().Error("live remote description failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err)
m.send(session, "Error", map[string]any{"code": "REMOTE_DESCRIPTION", "message": err.Error()})
return
}
answer, err := pc.CreateAnswer(nil)
if err != nil {
m.liveLogger().Error("live answer creation failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err)
m.send(session, "Error", map[string]any{"code": "CREATE_ANSWER", "message": err.Error()})
return
}
if err = pc.SetLocalDescription(answer); err != nil {
m.liveLogger().Error("live local description failed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "error", err)
m.send(session, "Error", map[string]any{"code": "LOCAL_DESCRIPTION", "message": err.Error()})
return
}
<-webrtc.GatheringCompletePromise(pc)
local := pc.LocalDescription()
if local == nil {
m.liveLogger().Error("live local description unavailable", "role", p.role, "stream_id", p.streamID, "user_id", p.userID)
m.send(session, "Error", map[string]any{"code": "NO_LOCAL_DESCRIPTION", "message": "local description unavailable"})
return
}
m.send(session, "Answer", map[string]any{"type": local.Type.String(), "sdp": local.SDP})
m.liveLogger().Info("live answer sent", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "sdp_bytes", len(local.SDP))
if p.role == "publisher" {
m.sendPublisherStatus(p.streamID, false, "waiting_for_tracks")
}
}
func (m *Manager) handleICEConnectionStateChange(p *peer, pc *webrtc.PeerConnection, state webrtc.ICEConnectionState) {
m.liveLogger().Info("live ICE state changed", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "state", state.String())
if state != webrtc.ICEConnectionStateFailed && state != webrtc.ICEConnectionStateClosed {
return
}
@@ -488,6 +521,7 @@ func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) {
return
}
entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), local: local}
m.liveLogger().Info("live publisher track received", "stream_id", p.streamID, "user_id", p.userID, "track_id", remote.ID(), "kind", strings.ToLower(remote.Kind().String()), "codec", codec.MimeType)
p.room.mu.Lock()
p.room.tracks[remote.Kind()] = entry
viewers := make([]*peer, 0, len(p.room.viewers))
@@ -513,6 +547,7 @@ func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) {
func (m *Manager) onDisconnect(session *melody.Session) {
if p := m.peer(session); p != nil {
m.liveLogger().Info("live signaling disconnected", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "remote_addr", sessionRemoteAddr(session))
m.closePeer(p, "disconnected")
return
}
@@ -558,6 +593,7 @@ func (m *Manager) closePeerConnection(p *peer, candidate *webrtc.PeerConnection,
return
}
p.cleanup.Do(func() {
m.liveLogger().Info("live peer closing", "role", p.role, "stream_id", p.streamID, "user_id", p.userID, "reason", reason)
if pc != nil {
_ = pc.Close()
}
@@ -721,15 +757,30 @@ func (m *Manager) sendPublisherStatusTo(session *melody.Session, streamID string
func (m *Manager) send(session *melody.Session, msgType string, payload any) {
encoded, err := json.Marshal(map[string]any{"type": msgType, "payload": payload})
if err != nil {
m.liveLogger().Error("live signaling response encoding failed", "message_type", msgType, "error", err)
return
}
_ = m.hub.Send(session, encoded)
if err = m.hub.Send(session, encoded); err != nil {
m.liveLogger().Warn("live signaling response send failed", "message_type", msgType, "remote_addr", sessionRemoteAddr(session), "error", err)
}
}
func (m *Manager) log(message string, args ...any) {
if m.logger != nil {
m.logger.Warn(message, args...)
m.liveLogger().Warn(message, args...)
}
func (m *Manager) liveLogger() *slog.Logger {
if m != nil && m.logger != nil {
return m.logger
}
return slog.Default().With("mod", "live")
}
func sessionRemoteAddr(session *melody.Session) string {
if session == nil || session.Request == nil {
return ""
}
return session.Request.RemoteAddr
}
func sessionUserID(session *melody.Session) (uint, bool) {
internal/logging/zap.go
+2
View File
@@ -435,6 +435,8 @@ func routedLogPaths(module string, level zapcore.Level) []string {
paths = append(paths, filepath.Join("timedTask", "task.log"))
case "websocket":
paths = append(paths, filepath.Join("websocket", "websocket.log"))
case "live":
paths = append(paths, filepath.Join("live", "live.log"))
case "error":
paths = append(paths, filepath.Join("error", "error.log"))
default:
internal/logging/zap_test.go
+2
View File
@@ -42,6 +42,7 @@ func TestZapHandlerRoutesHTTPAndErrorLogs(t *testing.T) {
logger := slog.New(handler)
logger.Info("request", "mod", "http", "request_id", "req-1")
logger.Info("handshake", "mod", "websocket")
logger.Info("publisher joined", "mod", "live")
logger.Error("failed", "mod", "users")
cleanup()
@@ -54,6 +55,7 @@ func TestZapHandlerRoutesHTTPAndErrorLogs(t *testing.T) {
filepath.Join(root, date, "application.log"),
filepath.Join(root, date, "http", "access.log"),
filepath.Join(root, date, "websocket", "websocket.log"),
filepath.Join(root, date, "live", "live.log"),
filepath.Join(root, date, "users", "application.log"),
filepath.Join(root, date, "error", "error.log"),
}