feat: live module and jenkins pipeline to harbor deploy
This commit is contained in:
Yvan
@@ -0,0 +1,5 @@
|
||||
package live
|
||||
|
||||
import "github.com/google/wire"
|
||||
|
||||
var ProviderSet = wire.NewSet(New)
|
||||
@@ -0,0 +1,714 @@
|
||||
package live
|
||||
|
||||
import (
|
||||
"context"
|
||||
"encoding/json"
|
||||
"errors"
|
||||
"fmt"
|
||||
"log/slog"
|
||||
"os"
|
||||
"strconv"
|
||||
"strings"
|
||||
"sync"
|
||||
"sync/atomic"
|
||||
"time"
|
||||
|
||||
bizgift "kra/internal/biz/gift"
|
||||
bizlive "kra/internal/biz/live"
|
||||
bizchat "kra/internal/biz/livechat"
|
||||
platformws "kra/pkg/websocket"
|
||||
|
||||
"github.com/olahol/melody"
|
||||
"github.com/pion/webrtc/v4"
|
||||
)
|
||||
|
||||
type Manager struct {
|
||||
hub platformws.Hub
|
||||
streams *bizlive.StreamUsecase
|
||||
roomCatalog *bizlive.RoomUsecase
|
||||
chat *bizchat.Usecase
|
||||
logger *slog.Logger
|
||||
api *webrtc.API
|
||||
mu sync.RWMutex
|
||||
rooms map[string]*room
|
||||
peers map[*melody.Session]*peer
|
||||
observers map[*melody.Session]struct{}
|
||||
nextSub atomic.Uint64
|
||||
stop chan struct{}
|
||||
stopOnce sync.Once
|
||||
}
|
||||
|
||||
type room struct {
|
||||
mu sync.RWMutex
|
||||
streamID string
|
||||
publisher *peer
|
||||
viewers map[*peer]struct{}
|
||||
tracks map[webrtc.RTPCodecType]*publishedTrack
|
||||
}
|
||||
|
||||
type publishedTrack struct {
|
||||
id string
|
||||
kind webrtc.RTPCodecType
|
||||
local *webrtc.TrackLocalStaticRTP
|
||||
}
|
||||
|
||||
type peer struct {
|
||||
session *melody.Session
|
||||
manager *Manager
|
||||
room *room
|
||||
streamID string
|
||||
userID uint
|
||||
role string
|
||||
subscriberID string
|
||||
pc *webrtc.PeerConnection
|
||||
lastBeat atomic.Int64
|
||||
cleanup sync.Once
|
||||
}
|
||||
|
||||
type message struct {
|
||||
Type string `json:"type"`
|
||||
ID string `json:"id,omitempty"`
|
||||
Payload json.RawMessage `json:"payload,omitempty"`
|
||||
}
|
||||
|
||||
type joinPayload struct {
|
||||
Role string `json:"role"`
|
||||
StreamID string `json:"streamId"`
|
||||
}
|
||||
|
||||
type sessionDescriptionPayload struct {
|
||||
Type string `json:"type"`
|
||||
SDP string `json:"sdp"`
|
||||
}
|
||||
|
||||
type chatSendPayload struct {
|
||||
Content string `json:"content"`
|
||||
}
|
||||
|
||||
func newPionAPI() (*webrtc.API, error) {
|
||||
settings := webrtc.SettingEngine{}
|
||||
minPort := envPort("KRA_LIVE_UDP_MIN_PORT")
|
||||
maxPort := envPort("KRA_LIVE_UDP_MAX_PORT")
|
||||
if minPort > 0 || maxPort > 0 {
|
||||
if minPort == 0 || maxPort == 0 || minPort > maxPort {
|
||||
return nil, fmt.Errorf("invalid KRA_LIVE_UDP_MIN_PORT/KRA_LIVE_UDP_MAX_PORT")
|
||||
}
|
||||
if err := settings.SetEphemeralUDPPortRange(minPort, maxPort); err != nil {
|
||||
return nil, fmt.Errorf("configure live udp port range: %w", err)
|
||||
}
|
||||
}
|
||||
return webrtc.NewAPI(webrtc.WithSettingEngine(settings)), nil
|
||||
}
|
||||
|
||||
func envPort(key string) uint16 {
|
||||
value, err := strconv.ParseUint(strings.TrimSpace(os.Getenv(key)), 10, 16)
|
||||
if err != nil {
|
||||
return 0
|
||||
}
|
||||
return uint16(value)
|
||||
}
|
||||
|
||||
func liveICEServers() []webrtc.ICEServer {
|
||||
value := strings.TrimSpace(os.Getenv("KRA_LIVE_STUN_URL"))
|
||||
if value == "" {
|
||||
return nil
|
||||
}
|
||||
return []webrtc.ICEServer{{URLs: []string{value}}}
|
||||
}
|
||||
|
||||
func New(hub platformws.Hub, streams *bizlive.StreamUsecase, rooms *bizlive.RoomUsecase, chat *bizchat.Usecase, logger *slog.Logger) (*Manager, func(), error) {
|
||||
if hub == nil {
|
||||
return nil, nil, errors.New("live websocket hub is unavailable")
|
||||
}
|
||||
if streams == nil {
|
||||
return nil, nil, errors.New("live stream usecase is unavailable")
|
||||
}
|
||||
if rooms == nil {
|
||||
return nil, nil, errors.New("live room usecase is unavailable")
|
||||
}
|
||||
if chat == nil {
|
||||
return nil, nil, errors.New("live chat usecase is unavailable")
|
||||
}
|
||||
api, err := newPionAPI()
|
||||
if err != nil {
|
||||
return nil, nil, err
|
||||
}
|
||||
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()
|
||||
return m, m.Close, nil
|
||||
}
|
||||
|
||||
func (m *Manager) Close() {
|
||||
m.stopOnce.Do(func() { close(m.stop) })
|
||||
m.mu.Lock()
|
||||
peers := make([]*peer, 0, len(m.peers))
|
||||
for _, p := range m.peers {
|
||||
peers = append(peers, p)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
for _, p := range peers {
|
||||
m.closePeer(p, "server_shutdown")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) onMessage(session *melody.Session, raw []byte) {
|
||||
var msg message
|
||||
if err := json.Unmarshal(raw, &msg); err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "INVALID_JSON", "message": "invalid json"})
|
||||
return
|
||||
}
|
||||
switch msg.Type {
|
||||
case "Join":
|
||||
var payload joinPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "invalid join payload"})
|
||||
return
|
||||
}
|
||||
m.join(session, payload)
|
||||
case "Offer":
|
||||
var payload sessionDescriptionPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "INVALID_OFFER", "message": "invalid offer payload"})
|
||||
return
|
||||
}
|
||||
m.handleOffer(session, payload)
|
||||
case "Heartbeat":
|
||||
if p := m.peer(session); p != nil {
|
||||
p.lastBeat.Store(time.Now().UnixMilli())
|
||||
}
|
||||
m.send(session, "HeartbeatAck", map[string]any{"streamId": peerStreamID(m, session)})
|
||||
case "Leave":
|
||||
if p := m.peer(session); p != nil {
|
||||
m.closePeer(p, "left")
|
||||
}
|
||||
case "ChatSend":
|
||||
var payload chatSendPayload
|
||||
if err := json.Unmarshal(msg.Payload, &payload); err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "INVALID_CHAT", "message": "invalid chat payload"})
|
||||
return
|
||||
}
|
||||
m.handleChat(session, payload)
|
||||
default:
|
||||
m.send(session, "Error", map[string]any{"code": "UNKNOWN_MESSAGE", "message": "unknown message type"})
|
||||
}
|
||||
}
|
||||
|
||||
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.mu.Lock()
|
||||
m.observers[session] = struct{}{}
|
||||
m.mu.Unlock()
|
||||
m.send(session, "Joined", map[string]any{"role": "observer"})
|
||||
m.sendStreamList(session)
|
||||
return
|
||||
}
|
||||
if streamID == "" || (role != "publisher" && role != "subscriber") {
|
||||
m.send(session, "JoinRejected", map[string]any{"code": "INVALID_JOIN", "message": "role and streamId are required"})
|
||||
return
|
||||
}
|
||||
userID, ok := sessionUserID(session)
|
||||
if !ok {
|
||||
m.send(session, "JoinRejected", map[string]any{"code": "AUTH_REQUIRED", "message": "authenticated user is required"})
|
||||
return
|
||||
}
|
||||
if existing := m.peer(session); existing != nil {
|
||||
m.closePeer(existing, "rejoin")
|
||||
}
|
||||
r := m.getOrCreateRoom(streamID)
|
||||
if role == "publisher" {
|
||||
if err := m.authorizeRoomPublisher(streamID, userID); err != nil {
|
||||
m.send(session, "JoinRejected", map[string]any{"code": "PUBLISH_FORBIDDEN", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
r.mu.RLock()
|
||||
occupied := r.publisher != nil
|
||||
r.mu.RUnlock()
|
||||
if occupied {
|
||||
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.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
|
||||
return
|
||||
}
|
||||
p := &peer{session: session, manager: m, room: r, streamID: streamID, userID: userID, role: role}
|
||||
p.lastBeat.Store(time.Now().UnixMilli())
|
||||
r.mu.Lock()
|
||||
r.publisher = p
|
||||
r.mu.Unlock()
|
||||
m.mu.Lock()
|
||||
m.peers[session] = p
|
||||
m.mu.Unlock()
|
||||
m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "mode": "pion-sfu"})
|
||||
m.sendChatHistory(session, streamID)
|
||||
m.sendPublisherStatus(streamID, false, "publisher_waiting_for_offer")
|
||||
m.broadcastStreamList()
|
||||
return
|
||||
}
|
||||
|
||||
// Match the old behavior: a user can only watch a room once. The old
|
||||
// connection is cleaned up before registering the new one.
|
||||
var old *peer
|
||||
r.mu.RLock()
|
||||
for viewer := range r.viewers {
|
||||
if viewer.userID == userID {
|
||||
old = viewer
|
||||
break
|
||||
}
|
||||
}
|
||||
r.mu.RUnlock()
|
||||
if old != nil {
|
||||
m.send(old.session, "Kicked", map[string]any{"reason": "logged_in_elsewhere", "streamId": streamID})
|
||||
m.closePeer(old, "logged_in_elsewhere")
|
||||
}
|
||||
if _, err := m.streams.RegisterSubscriber(context.Background(), streamID, userID); err != nil {
|
||||
m.send(session, "JoinRejected", map[string]any{"code": errorCode(err), "message": err.Error()})
|
||||
return
|
||||
}
|
||||
p := &peer{session: session, manager: m, room: r, streamID: streamID, userID: userID, role: role, subscriberID: "sub_" + strconv.FormatUint(m.nextSub.Add(1), 10)}
|
||||
p.lastBeat.Store(time.Now().UnixMilli())
|
||||
r.mu.Lock()
|
||||
r.viewers[p] = struct{}{}
|
||||
ready := r.publisher != nil && len(r.tracks) > 0
|
||||
r.mu.Unlock()
|
||||
m.mu.Lock()
|
||||
m.peers[session] = p
|
||||
m.mu.Unlock()
|
||||
m.send(session, "Joined", map[string]any{"role": role, "streamId": streamID, "subscriberId": p.subscriberID, "userId": userID, "mode": "pion-sfu"})
|
||||
m.sendChatHistory(session, streamID)
|
||||
m.sendPublisherStatusTo(session, streamID, ready, "publisher_offline")
|
||||
if ready {
|
||||
m.send(session, "PublisherStatus", map[string]any{"streamId": streamID, "ready": true})
|
||||
}
|
||||
if publisher := m.publisher(r); publisher != nil {
|
||||
m.send(publisher.session, "NewSubscriber", map[string]any{"streamId": streamID, "subscriberId": p.subscriberID, "userId": userID})
|
||||
}
|
||||
m.broadcastStreamList()
|
||||
}
|
||||
|
||||
func (m *Manager) handleChat(session *melody.Session, payload chatSendPayload) {
|
||||
p := m.peer(session)
|
||||
if p == nil || (p.role != "publisher" && p.role != "subscriber") {
|
||||
m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a room before chatting"})
|
||||
return
|
||||
}
|
||||
roomID, ok := roomIDFromStream(p.streamID)
|
||||
if !ok {
|
||||
m.send(session, "Error", map[string]any{"code": "CHAT_UNAVAILABLE", "message": "chat is only available in live rooms"})
|
||||
return
|
||||
}
|
||||
item, err := m.chat.Send(context.Background(), roomID, p.userID, payload.Content)
|
||||
if err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "INVALID_CHAT", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
m.broadcastRoom(p.streamID, "ChatMessage", chatMessagePayload(item))
|
||||
}
|
||||
|
||||
func (m *Manager) sendChatHistory(session *melody.Session, streamID string) {
|
||||
roomID, ok := roomIDFromStream(streamID)
|
||||
if !ok {
|
||||
return
|
||||
}
|
||||
items, err := m.chat.ListRecent(context.Background(), roomID, 100)
|
||||
if err != nil {
|
||||
m.log("load live chat history failed", "room_id", roomID, "error", err)
|
||||
return
|
||||
}
|
||||
messages := make([]map[string]any, 0, len(items))
|
||||
for _, item := range items {
|
||||
messages = append(messages, chatMessagePayload(item))
|
||||
}
|
||||
m.send(session, "ChatHistory", map[string]any{"roomId": roomID, "messages": messages})
|
||||
}
|
||||
|
||||
// GiftSent implements gift.Notifier. Gift persistence and wallet debit have
|
||||
// already committed; chat persistence is therefore best-effort and must never
|
||||
// turn a successful debit into a failed gift response.
|
||||
func (m *Manager) GiftSent(ctx context.Context, sent *bizgift.Send) {
|
||||
if sent == nil || sent.RoomID == nil || sent.GiftID == nil {
|
||||
return
|
||||
}
|
||||
item, err := m.chat.RecordGift(ctx, &bizchat.GiftSnapshot{RoomID: *sent.RoomID, UserID: sent.UserID, Username: sent.Username, Nickname: sent.Nickname, GiftID: *sent.GiftID, GiftName: sent.GiftName, GiftIcon: sent.GiftIcon, Quantity: sent.Quantity, DiamondTotal: sent.DiamondTotal})
|
||||
if err != nil {
|
||||
m.log("record live gift chat failed", "room_id", *sent.RoomID, "gift_send_id", sent.ID, "error", err)
|
||||
return
|
||||
}
|
||||
m.broadcastRoom("room-"+strconv.FormatUint(uint64(*sent.RoomID), 10), "GiftMessage", chatMessagePayload(item))
|
||||
}
|
||||
|
||||
func (m *Manager) broadcastRoom(streamID, msgType string, payload any) {
|
||||
m.mu.RLock()
|
||||
peers := make([]*peer, 0)
|
||||
for _, p := range m.peers {
|
||||
if p.streamID == streamID && (p.role == "publisher" || p.role == "subscriber") {
|
||||
peers = append(peers, p)
|
||||
}
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, p := range peers {
|
||||
m.send(p.session, msgType, payload)
|
||||
}
|
||||
}
|
||||
|
||||
func chatMessagePayload(item *bizchat.Message) map[string]any {
|
||||
return map[string]any{"id": item.ID, "roomId": item.RoomID, "type": item.Type, "userId": item.UserID, "username": item.Username, "nickname": item.Nickname, "avatar": item.Avatar, "content": item.Content, "giftId": item.GiftID, "giftName": item.GiftName, "giftIcon": item.GiftIcon, "quantity": item.Quantity, "diamondTotal": item.DiamondTotal, "createdAt": item.CreatedAt}
|
||||
}
|
||||
|
||||
func roomIDFromStream(streamID string) (uint, bool) {
|
||||
if !strings.HasPrefix(streamID, "room-") {
|
||||
return 0, false
|
||||
}
|
||||
value, err := strconv.ParseUint(strings.TrimPrefix(streamID, "room-"), 10, 64)
|
||||
return uint(value), err == nil && value > 0
|
||||
}
|
||||
|
||||
func (m *Manager) authorizeRoomPublisher(streamID string, userID uint) error {
|
||||
if !strings.HasPrefix(streamID, "room-") {
|
||||
return nil
|
||||
}
|
||||
id, err := strconv.ParseUint(strings.TrimPrefix(streamID, "room-"), 10, 64)
|
||||
if err != nil || id == 0 {
|
||||
return errors.New("invalid live room stream id")
|
||||
}
|
||||
item, err := m.roomCatalog.Get(context.Background(), uint(id))
|
||||
if err != nil {
|
||||
return errors.New("live room is unavailable")
|
||||
}
|
||||
if !item.Enabled {
|
||||
return errors.New("live room is disabled")
|
||||
}
|
||||
if item.AnchorUserID != userID {
|
||||
return errors.New("current user is not the room anchor")
|
||||
}
|
||||
return nil
|
||||
}
|
||||
|
||||
func (m *Manager) handleOffer(session *melody.Session, payload sessionDescriptionPayload) {
|
||||
p := m.peer(session)
|
||||
if p == nil || (p.role != "publisher" && p.role != "subscriber") {
|
||||
m.send(session, "Error", map[string]any{"code": "NOT_JOINED", "message": "join a stream first"})
|
||||
return
|
||||
}
|
||||
if strings.TrimSpace(payload.SDP) == "" {
|
||||
m.send(session, "Error", map[string]any{"code": "INVALID_SDP", "message": "sdp is required"})
|
||||
return
|
||||
}
|
||||
if p.pc != nil {
|
||||
_ = p.pc.Close()
|
||||
}
|
||||
pc, err := m.api.NewPeerConnection(webrtc.Configuration{ICEServers: liveICEServers()})
|
||||
if err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "PEER_CONNECTION", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
p.pc = pc
|
||||
pc.OnICEConnectionStateChange(func(state webrtc.ICEConnectionState) {
|
||||
if state == webrtc.ICEConnectionStateFailed || state == webrtc.ICEConnectionStateClosed {
|
||||
m.closePeer(p, "ice_"+state.String())
|
||||
}
|
||||
})
|
||||
if p.role == "publisher" {
|
||||
pc.OnTrack(func(track *webrtc.TrackRemote, _ *webrtc.RTPReceiver) {
|
||||
m.onPublisherTrack(p, track)
|
||||
})
|
||||
} else {
|
||||
for _, track := range m.trackSnapshot(p.room) {
|
||||
if sender, addErr := pc.AddTrack(track.local); addErr == nil {
|
||||
go drainRTCP(sender)
|
||||
}
|
||||
}
|
||||
}
|
||||
if err = pc.SetRemoteDescription(webrtc.SessionDescription{Type: webrtc.SDPTypeOffer, SDP: payload.SDP}); err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "REMOTE_DESCRIPTION", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
answer, err := pc.CreateAnswer(nil)
|
||||
if err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "CREATE_ANSWER", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
if err = pc.SetLocalDescription(answer); err != nil {
|
||||
m.send(session, "Error", map[string]any{"code": "LOCAL_DESCRIPTION", "message": err.Error()})
|
||||
return
|
||||
}
|
||||
<-webrtc.GatheringCompletePromise(pc)
|
||||
local := pc.LocalDescription()
|
||||
if local == nil {
|
||||
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})
|
||||
if p.role == "publisher" {
|
||||
m.sendPublisherStatus(p.streamID, false, "waiting_for_tracks")
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) onPublisherTrack(p *peer, remote *webrtc.TrackRemote) {
|
||||
codec := remote.Codec().RTPCodecCapability
|
||||
local, err := webrtc.NewTrackLocalStaticRTP(codec, remote.ID(), p.streamID)
|
||||
if err != nil {
|
||||
m.log("live track creation failed", "error", err)
|
||||
return
|
||||
}
|
||||
entry := &publishedTrack{id: remote.ID(), kind: remote.Kind(), local: local}
|
||||
p.room.mu.Lock()
|
||||
p.room.tracks[remote.Kind()] = entry
|
||||
viewers := make([]*peer, 0, len(p.room.viewers))
|
||||
for viewer := range p.room.viewers {
|
||||
viewers = append(viewers, viewer)
|
||||
}
|
||||
p.room.mu.Unlock()
|
||||
m.sendPublisherStatus(p.streamID, true, "")
|
||||
for _, viewer := range viewers {
|
||||
m.send(viewer.session, "NewProducer", map[string]any{"streamId": p.streamID, "producerId": remote.ID(), "kind": strings.ToLower(remote.Kind().String())})
|
||||
m.send(viewer.session, "Renegotiate", map[string]any{"streamId": p.streamID})
|
||||
}
|
||||
go func() {
|
||||
for {
|
||||
packet, _, readErr := remote.ReadRTP()
|
||||
if readErr != nil {
|
||||
return
|
||||
}
|
||||
_ = local.WriteRTP(packet)
|
||||
}
|
||||
}()
|
||||
}
|
||||
|
||||
func (m *Manager) onDisconnect(session *melody.Session) {
|
||||
if p := m.peer(session); p != nil {
|
||||
m.closePeer(p, "disconnected")
|
||||
return
|
||||
}
|
||||
m.mu.Lock()
|
||||
delete(m.observers, session)
|
||||
m.mu.Unlock()
|
||||
}
|
||||
|
||||
func (m *Manager) heartbeatLoop() {
|
||||
ticker := time.NewTicker(10 * time.Second)
|
||||
defer ticker.Stop()
|
||||
for {
|
||||
select {
|
||||
case <-ticker.C:
|
||||
now := time.Now().UnixMilli()
|
||||
m.mu.RLock()
|
||||
peers := make([]*peer, 0, len(m.peers))
|
||||
for _, p := range m.peers {
|
||||
peers = append(peers, p)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, p := range peers {
|
||||
if now-p.lastBeat.Load() >= (45 * time.Second).Milliseconds() {
|
||||
m.closePeer(p, "heartbeat_timeout")
|
||||
}
|
||||
}
|
||||
case <-m.stop:
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) closePeer(p *peer, reason string) {
|
||||
if p == nil {
|
||||
return
|
||||
}
|
||||
p.cleanup.Do(func() {
|
||||
if p.pc != nil {
|
||||
_ = p.pc.Close()
|
||||
}
|
||||
if p.role == "publisher" {
|
||||
_ = m.streams.UnregisterPublisher(context.Background(), p.streamID, p.userID)
|
||||
p.room.mu.Lock()
|
||||
tracks := make([]*publishedTrack, 0, len(p.room.tracks))
|
||||
for _, track := range p.room.tracks {
|
||||
tracks = append(tracks, track)
|
||||
}
|
||||
if p.room.publisher == p {
|
||||
p.room.publisher = nil
|
||||
}
|
||||
p.room.tracks = make(map[webrtc.RTPCodecType]*publishedTrack)
|
||||
viewers := make([]*peer, 0, len(p.room.viewers))
|
||||
for viewer := range p.room.viewers {
|
||||
viewers = append(viewers, viewer)
|
||||
}
|
||||
p.room.mu.Unlock()
|
||||
for _, viewer := range viewers {
|
||||
for _, track := range tracks {
|
||||
m.send(viewer.session, "ProducerClosed", map[string]any{"streamId": p.streamID, "producerId": track.id, "kind": strings.ToLower(track.kind.String()), "reason": reason})
|
||||
}
|
||||
m.send(viewer.session, "PublisherStatus", map[string]any{"streamId": p.streamID, "ready": false, "reason": reason})
|
||||
}
|
||||
} else if p.role == "subscriber" {
|
||||
_ = m.streams.UnregisterSubscriber(context.Background(), p.streamID, p.userID)
|
||||
p.room.mu.Lock()
|
||||
delete(p.room.viewers, p)
|
||||
publisher := p.room.publisher
|
||||
p.room.mu.Unlock()
|
||||
if publisher != nil {
|
||||
m.send(publisher.session, "SubscriberLeft", map[string]any{"streamId": p.streamID, "subscriberId": p.subscriberID})
|
||||
}
|
||||
}
|
||||
m.mu.Lock()
|
||||
delete(m.peers, p.session)
|
||||
p.room.mu.RLock()
|
||||
emptyRoom := p.room.publisher == nil && len(p.room.viewers) == 0
|
||||
p.room.mu.RUnlock()
|
||||
if emptyRoom {
|
||||
delete(m.rooms, p.streamID)
|
||||
}
|
||||
m.mu.Unlock()
|
||||
m.broadcastStreamList()
|
||||
})
|
||||
}
|
||||
|
||||
func (m *Manager) getOrCreateRoom(streamID string) *room {
|
||||
m.mu.Lock()
|
||||
defer m.mu.Unlock()
|
||||
if r := m.rooms[streamID]; r != nil {
|
||||
return r
|
||||
}
|
||||
r := &room{streamID: streamID, viewers: make(map[*peer]struct{}), tracks: make(map[webrtc.RTPCodecType]*publishedTrack)}
|
||||
m.rooms[streamID] = r
|
||||
return r
|
||||
}
|
||||
|
||||
func (m *Manager) peer(session *melody.Session) *peer {
|
||||
m.mu.RLock()
|
||||
defer m.mu.RUnlock()
|
||||
return m.peers[session]
|
||||
}
|
||||
|
||||
func (m *Manager) publisher(r *room) *peer {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
return r.publisher
|
||||
}
|
||||
|
||||
func (m *Manager) trackSnapshot(r *room) []*publishedTrack {
|
||||
r.mu.RLock()
|
||||
defer r.mu.RUnlock()
|
||||
tracks := make([]*publishedTrack, 0, len(r.tracks))
|
||||
for _, track := range r.tracks {
|
||||
tracks = append(tracks, track)
|
||||
}
|
||||
return tracks
|
||||
}
|
||||
|
||||
func (m *Manager) sendStreamList(session *melody.Session) {
|
||||
streams, err := m.streams.List(context.Background())
|
||||
if err != nil {
|
||||
return
|
||||
}
|
||||
items := make([]map[string]any, 0, len(streams))
|
||||
for _, stream := range streams {
|
||||
items = append(items, map[string]any{"streamId": stream.StreamID, "publisherOnline": stream.PublisherOnline, "subscriberCount": stream.SubscriberCount})
|
||||
}
|
||||
m.send(session, "StreamListUpdate", map[string]any{"streams": items})
|
||||
}
|
||||
|
||||
func (m *Manager) broadcastStreamList() {
|
||||
m.mu.RLock()
|
||||
observers := make([]*melody.Session, 0, len(m.observers))
|
||||
for session := range m.observers {
|
||||
observers = append(observers, session)
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, session := range observers {
|
||||
m.sendStreamList(session)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) sendPublisherStatus(streamID string, ready bool, reason string) {
|
||||
m.mu.RLock()
|
||||
peers := make([]*peer, 0, len(m.peers))
|
||||
for _, p := range m.peers {
|
||||
if p.role == "subscriber" && p.streamID == streamID {
|
||||
peers = append(peers, p)
|
||||
}
|
||||
}
|
||||
m.mu.RUnlock()
|
||||
for _, p := range peers {
|
||||
m.sendPublisherStatusTo(p.session, streamID, ready, reason)
|
||||
}
|
||||
}
|
||||
|
||||
func (m *Manager) sendPublisherStatusTo(session *melody.Session, streamID string, ready bool, reason string) {
|
||||
payload := map[string]any{"streamId": streamID, "ready": ready}
|
||||
if reason != "" {
|
||||
payload["reason"] = reason
|
||||
}
|
||||
m.send(session, "PublisherStatus", payload)
|
||||
}
|
||||
|
||||
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 {
|
||||
return
|
||||
}
|
||||
_ = m.hub.Send(session, encoded)
|
||||
}
|
||||
|
||||
func (m *Manager) log(message string, args ...any) {
|
||||
if m.logger != nil {
|
||||
m.logger.Warn(message, args...)
|
||||
}
|
||||
}
|
||||
|
||||
func sessionUserID(session *melody.Session) (uint, bool) {
|
||||
if session == nil {
|
||||
return 0, false
|
||||
}
|
||||
value, ok := session.Keys["user_id"]
|
||||
if !ok || value == nil {
|
||||
return 0, false
|
||||
}
|
||||
switch typed := value.(type) {
|
||||
case uint:
|
||||
return typed, typed > 0
|
||||
case uint64:
|
||||
return uint(typed), typed > 0
|
||||
case int:
|
||||
return uint(typed), typed > 0
|
||||
case int64:
|
||||
return uint(typed), typed > 0
|
||||
case string:
|
||||
parsed, err := strconv.ParseUint(typed, 10, 64)
|
||||
return uint(parsed), err == nil && parsed > 0
|
||||
default:
|
||||
parsed, err := strconv.ParseUint(fmt.Sprint(value), 10, 64)
|
||||
return uint(parsed), err == nil && parsed > 0
|
||||
}
|
||||
}
|
||||
|
||||
func peerStreamID(m *Manager, session *melody.Session) string {
|
||||
if p := m.peer(session); p != nil {
|
||||
return p.streamID
|
||||
}
|
||||
return ""
|
||||
}
|
||||
|
||||
func errorCode(err error) string {
|
||||
switch {
|
||||
case errors.Is(err, bizlive.ErrMaxStreams):
|
||||
return "MAX_STREAMS"
|
||||
case errors.Is(err, bizlive.ErrPublisherExists):
|
||||
return "DUPLICATE_PUBLISHER"
|
||||
case errors.Is(err, bizlive.ErrSubscriberExists):
|
||||
return "DUPLICATE_SUBSCRIBER"
|
||||
default:
|
||||
return "INVALID_JOIN"
|
||||
}
|
||||
}
|
||||
|
||||
func drainRTCP(sender *webrtc.RTPSender) {
|
||||
for {
|
||||
if _, _, err := sender.ReadRTCP(); err != nil {
|
||||
return
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -4,11 +4,13 @@
|
||||
package integration
|
||||
|
||||
import (
|
||||
bizgift "kra/internal/biz/gift"
|
||||
integrationbiz "kra/internal/biz/integration"
|
||||
bizpayment "kra/internal/biz/payment"
|
||||
"kra/internal/biz/system"
|
||||
"kra/internal/integration/cache"
|
||||
"kra/internal/integration/email"
|
||||
liveintegration "kra/internal/integration/live"
|
||||
mqintegration "kra/internal/integration/mq"
|
||||
paymentintegration "kra/internal/integration/payment"
|
||||
"kra/internal/integration/storage"
|
||||
@@ -36,5 +38,7 @@ var ProviderSet = wire.NewSet(
|
||||
wire.Bind(new(mq.SubscriptionRegistrar), new(*mqintegration.Reloadable)),
|
||||
websocketintegration.New,
|
||||
wire.Bind(new(platformws.Hub), new(*websocketintegration.Server)),
|
||||
liveintegration.ProviderSet,
|
||||
wire.Bind(new(bizgift.Notifier), new(*liveintegration.Manager)),
|
||||
wire.Bind(new(system.FileStorage), new(*storage.Reloadable)),
|
||||
)
|
||||
|
||||
Reference in New Issue
Block a user