feat: sync group call livestream support
This commit is contained in:
A
parent
56d995474c
commit
f1a27996d3
37 changed files with 2219 additions and 83 deletions
|
|
@ -6,9 +6,12 @@ package groupcalls
|
|||
import (
|
||||
"context"
|
||||
"crypto/rand"
|
||||
"crypto/subtle"
|
||||
"encoding/base64"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"strconv"
|
||||
"strings"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
"telesrv/internal/store"
|
||||
|
|
@ -24,8 +27,10 @@ func NewService(st store.GroupCallStore) *Service {
|
|||
return &Service{store: st}
|
||||
}
|
||||
|
||||
// Create 分配 id/access_hash 并建会。
|
||||
func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, title string, now int) (domain.GroupCall, error) {
|
||||
// Create 分配 id/access_hash 并建会。rtmpStream=true 创建 RTMP 直播房间；
|
||||
// joinMuted=true（广播频道直播）让非管理员入会即被静音且不可自解；
|
||||
// scheduleDate>0 创建定时通话（客户端倒计时等待 startScheduled）。
|
||||
func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, title string, rtmpStream, joinMuted bool, scheduleDate, now int) (domain.GroupCall, error) {
|
||||
id, err := randomPositiveInt64()
|
||||
if err != nil {
|
||||
return domain.GroupCall{}, err
|
||||
|
|
@ -40,11 +45,73 @@ func (s *Service) Create(ctx context.Context, channelID, creatorUserID int64, ti
|
|||
ChannelID: channelID,
|
||||
CreatorUserID: creatorUserID,
|
||||
Title: title,
|
||||
RtmpStream: rtmpStream,
|
||||
JoinMuted: joinMuted,
|
||||
ScheduleDate: scheduleDate,
|
||||
Version: 1,
|
||||
CreatedAt: now,
|
||||
})
|
||||
}
|
||||
|
||||
// StartScheduled 把定时通话转为进行中（清 schedule_date）；changed=false 幂等。
|
||||
func (s *Service) StartScheduled(ctx context.Context, callID int64) (domain.GroupCall, bool, error) {
|
||||
return s.store.StartScheduledGroupCall(ctx, callID)
|
||||
}
|
||||
|
||||
// SetScheduleSubscription 写入/清除开播提醒订阅。
|
||||
func (s *Service) SetScheduleSubscription(ctx context.Context, callID, userID int64, subscribed bool) error {
|
||||
return s.store.SetScheduleStartSubscription(ctx, callID, userID, subscribed)
|
||||
}
|
||||
|
||||
// ScheduleSubscriberIDs 返回订阅开播提醒的 userID。
|
||||
func (s *Service) ScheduleSubscriberIDs(ctx context.Context, callID int64) ([]int64, error) {
|
||||
return s.store.ListScheduleSubscriberIDs(ctx, callID)
|
||||
}
|
||||
|
||||
// RtmpStreamKey 返回 channel 的持久 RTMP 推流密钥；不存在或 rotate=true 时生成
|
||||
// 新 key（覆盖写入，旧 key 即刻失效）。key 形如 "<channelID>_<hex>"，ingest 端
|
||||
// 据前缀定位 channel、再整串比对鉴权。
|
||||
func (s *Service) RtmpStreamKey(ctx context.Context, channelID int64, rotate bool, now int) (string, error) {
|
||||
if !rotate {
|
||||
key, found, err := s.store.GetRtmpStreamKey(ctx, channelID)
|
||||
if err != nil {
|
||||
return "", err
|
||||
}
|
||||
if found {
|
||||
return key, nil
|
||||
}
|
||||
}
|
||||
var buf [24]byte
|
||||
if _, err := rand.Read(buf[:]); err != nil {
|
||||
return "", fmt.Errorf("groupcalls: random rtmp key: %w", err)
|
||||
}
|
||||
key := fmt.Sprintf("%d_%x", channelID, buf)
|
||||
if err := s.store.SetRtmpStreamKey(ctx, channelID, key, now); err != nil {
|
||||
return "", err
|
||||
}
|
||||
return key, nil
|
||||
}
|
||||
|
||||
// VerifyRtmpStreamKey 校验推流密钥并返回其所属 channelID（RTMP ingest 鉴权入口）。
|
||||
func (s *Service) VerifyRtmpStreamKey(ctx context.Context, key string) (int64, bool, error) {
|
||||
sep := strings.IndexByte(key, '_')
|
||||
if sep <= 0 {
|
||||
return 0, false, nil
|
||||
}
|
||||
channelID, err := strconv.ParseInt(key[:sep], 10, 64)
|
||||
if err != nil || channelID <= 0 {
|
||||
return 0, false, nil
|
||||
}
|
||||
stored, found, err := s.store.GetRtmpStreamKey(ctx, channelID)
|
||||
if err != nil {
|
||||
return 0, false, err
|
||||
}
|
||||
if !found || subtle.ConstantTimeCompare([]byte(stored), []byte(key)) != 1 {
|
||||
return 0, false, nil
|
||||
}
|
||||
return channelID, true, nil
|
||||
}
|
||||
|
||||
// CreateConference 分配 id/access_hash/slug 并创建 ad-hoc conference call。
|
||||
func (s *Service) CreateConference(ctx context.Context, creatorUserID, randomID, migratedFromPhoneCallID int64, now int) (domain.GroupCall, error) {
|
||||
for i := 0; i < 8; i++ {
|
||||
|
|
|
|||
70
internal/app/livestream/part.go
Normal file
70
internal/app/livestream/part.go
Normal file
|
|
@ -0,0 +1,70 @@
|
|||
package livestream
|
||||
|
||||
import "encoding/binary"
|
||||
|
||||
// tgcalls broadcast part 打包（消费方 tgcalls VideoStreamingPart.cpp
|
||||
// consumeVideoStreamInfo）。unified（RTMP）模式的 part 结构：
|
||||
//
|
||||
// int32(LE) 签名 0xa12e810d
|
||||
// TL 风格短字符串：容器名（本实现恒 "mp4"）
|
||||
// int32 activeMask（单一 unified 轨恒 1）
|
||||
// int32 eventCount（消费方只读第一个 event，恒写 1）
|
||||
// event: int32 offset(=0，相对头部之后的数据) + 字符串 endpointId("unified")
|
||||
// + int32 rotation(0) + int32 extra(0)
|
||||
// 随后紧跟容器数据（音视频同容器，客户端分别按 Video/Audio content type 解）。
|
||||
const partSignature uint32 = 0xa12e810d
|
||||
|
||||
// partContainer 必须落在 TDesktop 裁剪版 ffmpeg 的 demuxer 白名单内
|
||||
// （Telegram/build/prepare/prepare.py --enable-demuxer=...，含 mov/mp4、无 mpegts）。
|
||||
// "mp4" 由 mov demuxer 别名匹配；tgcalls AVIO 支持 seek，完整 mp4（moov 在尾）可解。
|
||||
const (
|
||||
partContainer = "mp4"
|
||||
partEndpointID = "unified"
|
||||
)
|
||||
|
||||
// appendTLString 按 tgcalls readSerializedString 的逆操作写入字符串：
|
||||
// 长度 <254 用 1 字节长度 + 数据，整体（含长度字节）补齐到 4 字节；
|
||||
// 否则 0xFE + 3 字节小端长度 + 数据，数据补齐到 4 字节。
|
||||
func appendTLString(dst []byte, s string) []byte {
|
||||
n := len(s)
|
||||
if n < 254 {
|
||||
dst = append(dst, byte(n))
|
||||
dst = append(dst, s...)
|
||||
for (n+1)%4 != 0 {
|
||||
dst = append(dst, 0)
|
||||
n++
|
||||
}
|
||||
return dst
|
||||
}
|
||||
dst = append(dst, 0xFE, byte(n), byte(n>>8), byte(n>>16))
|
||||
dst = append(dst, s...)
|
||||
for n%4 != 0 {
|
||||
dst = append(dst, 0)
|
||||
n++
|
||||
}
|
||||
return dst
|
||||
}
|
||||
|
||||
func appendUint32(dst []byte, v uint32) []byte {
|
||||
var buf [4]byte
|
||||
binary.LittleEndian.PutUint32(buf[:], v)
|
||||
return append(dst, buf[:]...)
|
||||
}
|
||||
|
||||
func appendInt32(dst []byte, v int32) []byte {
|
||||
return appendUint32(dst, uint32(v))
|
||||
}
|
||||
|
||||
// packUnifiedPart 把一段自包含 MPEG-TS 数据包成 tgcalls broadcast part。
|
||||
func packUnifiedPart(tsData []byte) []byte {
|
||||
out := make([]byte, 0, len(tsData)+48)
|
||||
out = appendUint32(out, partSignature)
|
||||
out = appendTLString(out, partContainer)
|
||||
out = appendInt32(out, 1) // activeMask
|
||||
out = appendInt32(out, 1) // eventCount
|
||||
out = appendInt32(out, 0) // event.offset
|
||||
out = appendTLString(out, partEndpointID)
|
||||
out = appendInt32(out, 0) // event.rotation
|
||||
out = appendInt32(out, 0) // event.extra
|
||||
return append(out, tsData...)
|
||||
}
|
||||
103
internal/app/livestream/part_test.go
Normal file
103
internal/app/livestream/part_test.go
Normal file
|
|
@ -0,0 +1,103 @@
|
|||
package livestream
|
||||
|
||||
import (
|
||||
"encoding/binary"
|
||||
"testing"
|
||||
)
|
||||
|
||||
// readTLString 复刻 tgcalls readSerializedString，用于反解 packUnifiedPart 的头。
|
||||
func readTLString(data []byte, off *int) (string, bool) {
|
||||
if *off >= len(data) {
|
||||
return "", false
|
||||
}
|
||||
first := int(data[*off])
|
||||
*off++
|
||||
var length, padding int
|
||||
if first == 254 {
|
||||
if *off+3 > len(data) {
|
||||
return "", false
|
||||
}
|
||||
length = int(data[*off]) | int(data[*off+1])<<8 | int(data[*off+2])<<16
|
||||
*off += 3
|
||||
padding = (4 - length%4) % 4
|
||||
} else {
|
||||
length = first
|
||||
padding = (4 - (length+1)%4) % 4
|
||||
}
|
||||
if *off+length > len(data) {
|
||||
return "", false
|
||||
}
|
||||
s := string(data[*off : *off+length])
|
||||
*off += length + padding
|
||||
return s, true
|
||||
}
|
||||
|
||||
func readI32(data []byte, off *int) (int32, bool) {
|
||||
if *off+4 > len(data) {
|
||||
return 0, false
|
||||
}
|
||||
v := int32(binary.LittleEndian.Uint32(data[*off : *off+4]))
|
||||
*off += 4
|
||||
return v, true
|
||||
}
|
||||
|
||||
// TestPackUnifiedPartMatchesTgcallsHeader 校验打包头逐字段可被 tgcalls
|
||||
// consumeVideoStreamInfo 解出：签名、容器名、activeMask、单 event(endpoint="unified")，
|
||||
// 且 event.offset=0 对应紧随其后的 TS 数据。
|
||||
func TestPackUnifiedPartMatchesTgcallsHeader(t *testing.T) {
|
||||
ts := []byte{0x47, 0x40, 0x00, 0x10, 0xDE, 0xAD, 0xBE, 0xEF} // 伪 TS 数据
|
||||
part := packUnifiedPart(ts)
|
||||
|
||||
off := 0
|
||||
sig, ok := readI32(part, &off)
|
||||
if !ok || uint32(sig) != partSignature {
|
||||
t.Fatalf("signature = %#x ok=%v, want %#x", uint32(sig), ok, partSignature)
|
||||
}
|
||||
container, ok := readTLString(part, &off)
|
||||
if !ok || container != partContainer {
|
||||
t.Fatalf("container = %q ok=%v, want %q", container, ok, partContainer)
|
||||
}
|
||||
activeMask, ok := readI32(part, &off)
|
||||
if !ok || activeMask != 1 {
|
||||
t.Fatalf("activeMask = %d ok=%v, want 1", activeMask, ok)
|
||||
}
|
||||
eventCount, ok := readI32(part, &off)
|
||||
if !ok || eventCount != 1 {
|
||||
t.Fatalf("eventCount = %d ok=%v, want 1", eventCount, ok)
|
||||
}
|
||||
eventOffset, ok := readI32(part, &off)
|
||||
if !ok || eventOffset != 0 {
|
||||
t.Fatalf("event.offset = %d ok=%v, want 0", eventOffset, ok)
|
||||
}
|
||||
endpoint, ok := readTLString(part, &off)
|
||||
if !ok || endpoint != partEndpointID {
|
||||
t.Fatalf("endpoint = %q ok=%v, want %q", endpoint, ok, partEndpointID)
|
||||
}
|
||||
rotation, _ := readI32(part, &off)
|
||||
extra, _ := readI32(part, &off)
|
||||
if rotation != 0 || extra != 0 {
|
||||
t.Fatalf("rotation/extra = %d/%d, want 0/0", rotation, extra)
|
||||
}
|
||||
// 头之后（event.offset=0 起）应为原始 TS 数据。
|
||||
if got := part[off:]; string(got) != string(ts) {
|
||||
t.Fatalf("payload = %x, want %x", got, ts)
|
||||
}
|
||||
}
|
||||
|
||||
// TestAppendTLStringPadding 校验短/长字符串都补齐到 4 字节边界。
|
||||
func TestAppendTLStringPadding(t *testing.T) {
|
||||
for _, s := range []string{"", "a", "ab", "abc", "mpegts", "unified"} {
|
||||
out := appendTLString(nil, s)
|
||||
if len(out)%4 != 0 {
|
||||
t.Fatalf("appendTLString(%q) len=%d not 4-aligned", s, len(out))
|
||||
}
|
||||
off := 0
|
||||
got, ok := readTLString(out, &off)
|
||||
if !ok || got != s {
|
||||
t.Fatalf("roundtrip %q -> %q ok=%v", s, got, ok)
|
||||
}
|
||||
if off != len(out) {
|
||||
t.Fatalf("roundtrip %q consumed %d of %d bytes", s, off, len(out))
|
||||
}
|
||||
}
|
||||
}
|
||||
68
internal/app/livestream/rtmp.go
Normal file
68
internal/app/livestream/rtmp.go
Normal file
|
|
@ -0,0 +1,68 @@
|
|||
package livestream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"io"
|
||||
"net"
|
||||
|
||||
rtmp "github.com/yutopp/go-rtmp"
|
||||
rtmpmsg "github.com/yutopp/go-rtmp/message"
|
||||
"go.uber.org/zap"
|
||||
)
|
||||
|
||||
// rtmpHandler 是单条 RTMP 连接的回调：publish 时用 stream key 鉴权并绑定
|
||||
// channel，media tag 直通 FLV → ffmpeg。一条连接只允许一路 publish。
|
||||
type rtmpHandler struct {
|
||||
rtmp.DefaultHandler
|
||||
svc *Service
|
||||
conn net.Conn
|
||||
stream *stream
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) OnPublish(_ *rtmp.StreamContext, _ uint32, cmd *rtmpmsg.NetStreamPublish) error {
|
||||
if h.stream != nil {
|
||||
return errPublishRejected
|
||||
}
|
||||
st, err := h.svc.startPublish(context.Background(), cmd.PublishingName, h.conn)
|
||||
if err != nil {
|
||||
h.svc.log.Warn("rtmp publish rejected", zap.Error(err))
|
||||
return errPublishRejected
|
||||
}
|
||||
h.stream = st
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) OnSetDataFrame(timestamp uint32, data *rtmpmsg.NetStreamSetDataFrame) error {
|
||||
if h.stream == nil {
|
||||
return nil
|
||||
}
|
||||
// onMetaData 原样透传给 ffmpeg（可选信息，写失败不断流）。
|
||||
_ = h.stream.writeTag(18, timestamp, data.Payload)
|
||||
return nil
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) OnAudio(timestamp uint32, payload io.Reader) error {
|
||||
return h.writeMedia(8, timestamp, payload)
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) OnVideo(timestamp uint32, payload io.Reader) error {
|
||||
return h.writeMedia(9, timestamp, payload)
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) writeMedia(tagType byte, timestamp uint32, payload io.Reader) error {
|
||||
if h.stream == nil {
|
||||
return nil
|
||||
}
|
||||
body, err := io.ReadAll(payload)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
return h.stream.writeTag(tagType, timestamp, body)
|
||||
}
|
||||
|
||||
func (h *rtmpHandler) OnClose() {
|
||||
if h.stream != nil {
|
||||
h.svc.endPublish(h.stream)
|
||||
h.stream = nil
|
||||
}
|
||||
}
|
||||
285
internal/app/livestream/segmenter.go
Normal file
285
internal/app/livestream/segmenter.go
Normal file
|
|
@ -0,0 +1,285 @@
|
|||
package livestream
|
||||
|
||||
import (
|
||||
"bufio"
|
||||
"encoding/binary"
|
||||
"fmt"
|
||||
"io"
|
||||
"os"
|
||||
"os/exec"
|
||||
"path/filepath"
|
||||
"sync"
|
||||
|
||||
"go.uber.org/zap"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
// segmentDurationMs 是 broadcast part 的固定时长。tgcalls StreamingMediaContext
|
||||
// 写死 _segmentDuration=1000（scale 0），时间轴推进按 +1000 走，segment 必须严格
|
||||
// 1 秒切齐（转码强制每秒关键帧保证切点）。
|
||||
const segmentDurationMs = 1000
|
||||
|
||||
// minSegmentsBeforeAnnounce：客户端拿到 last_timestamp_ms 后从 last-2000 开始拉，
|
||||
// 至少积 3 段再对外公布时间轴，避免起播即请求不存在的负偏移段。
|
||||
const minSegmentsBeforeAnnounce = 3
|
||||
|
||||
// stream 是一路活跃 RTMP 推流：FLV 入 ffmpeg（转码+按秒切 MPEG-TS）→ 打包 part
|
||||
// 入内存 ring。时间轴：T0 取首段完成时刻向下取整秒，第 i 段的 time_ms = T0+i*1000。
|
||||
type stream struct {
|
||||
channelID int64
|
||||
log *zap.Logger
|
||||
dir string
|
||||
keep int
|
||||
|
||||
cmd *exec.Cmd
|
||||
stdin io.WriteCloser
|
||||
closer io.Closer // RTMP 连接，DropChannel 时踢掉推流端
|
||||
|
||||
mu sync.Mutex
|
||||
flvStarted bool
|
||||
baseMs int64 // T0；0=尚未产出任何 segment
|
||||
segments map[int64][]byte // time_ms → packed part
|
||||
order []int64 // 按 time_ms 升序（ring 淘汰用）
|
||||
lastMs int64 // 最新 segment 的 time_ms
|
||||
segmentSeq int64 // 已完成 segment 计数
|
||||
ended bool
|
||||
oversizeWas bool
|
||||
|
||||
nowMs func() int64
|
||||
}
|
||||
|
||||
// ffmpegArgs 组装转码+切段命令。要点：
|
||||
// - 强制每秒关键帧（-force_key_frames）保证 -f segment 严格按 1s 切；
|
||||
// - 严格码率上限：TDesktop 拉 part 单次 `upload.getFile(offset=0,limit=128KiB)`
|
||||
// 且**不续读**，单段（视频+音频+TS 开销）>128KiB 会被静默截断致花屏。
|
||||
// 故 unified 单质量必须压在 ~1Mbps 以下——这里目标 ~640kbps：视频
|
||||
// 480k（maxrate/bufsize=480k 收紧到每秒 VBV，杜绝关键帧段爆量）+ 音频 64k，
|
||||
// 并降到 640x360/24fps 进一步留余量；
|
||||
// - 输出自包含 mp4（每段独立 moov，可单独 avformat_open_input）。⚠ 不能用
|
||||
// MPEG-TS：TDesktop 裁剪版 ffmpeg 的 demuxer 白名单只有 mov/mp4 系
|
||||
// （prepare.py --enable-demuxer），mpegts 会让 tgcalls 打不开容器 → 黑屏；
|
||||
// - -segment_list pipe:1 每完成一段输出一行文件名，作为完成事件。
|
||||
func ffmpegArgs(outDir string) []string {
|
||||
return []string{
|
||||
"-hide_banner", "-nostats", "-loglevel", "warning",
|
||||
"-fflags", "+genpts",
|
||||
"-f", "flv", "-i", "pipe:0",
|
||||
"-vf", "scale=-2:360", "-r", "24",
|
||||
"-c:v", "libx264", "-preset", "veryfast", "-tune", "zerolatency",
|
||||
"-profile:v", "main", "-pix_fmt", "yuv420p",
|
||||
"-b:v", "480k", "-maxrate", "480k", "-bufsize", "480k",
|
||||
"-g", "24", "-keyint_min", "24",
|
||||
"-force_key_frames", "expr:gte(t,n_forced*1)", "-sc_threshold", "0",
|
||||
"-c:a", "aac", "-b:a", "64k", "-ar", "48000", "-ac", "2",
|
||||
"-f", "segment",
|
||||
"-segment_time", "1",
|
||||
"-segment_format", "mp4",
|
||||
"-segment_format_options", "movflags=+faststart",
|
||||
"-segment_list", "pipe:1",
|
||||
"-segment_list_type", "flat",
|
||||
"-reset_timestamps", "1",
|
||||
filepath.Join(outDir, "seg%06d.mp4"),
|
||||
}
|
||||
}
|
||||
|
||||
func newStream(channelID int64, ffmpegPath, workDir string, keep int, closer io.Closer, nowMs func() int64, log *zap.Logger) (*stream, error) {
|
||||
dir, err := os.MkdirTemp(workDir, fmt.Sprintf("live_%d_", channelID))
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("livestream: workdir: %w", err)
|
||||
}
|
||||
cmd := exec.Command(ffmpegPath, ffmpegArgs(dir)...)
|
||||
stdin, err := cmd.StdinPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("livestream: ffmpeg stdin: %w", err)
|
||||
}
|
||||
stdout, err := cmd.StdoutPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("livestream: ffmpeg stdout: %w", err)
|
||||
}
|
||||
stderr, err := cmd.StderrPipe()
|
||||
if err != nil {
|
||||
return nil, fmt.Errorf("livestream: ffmpeg stderr: %w", err)
|
||||
}
|
||||
if err := cmd.Start(); err != nil {
|
||||
return nil, fmt.Errorf("livestream: start ffmpeg: %w", err)
|
||||
}
|
||||
s := &stream{
|
||||
channelID: channelID,
|
||||
log: log,
|
||||
dir: dir,
|
||||
keep: keep,
|
||||
cmd: cmd,
|
||||
stdin: stdin,
|
||||
closer: closer,
|
||||
segments: make(map[int64][]byte),
|
||||
nowMs: nowMs,
|
||||
}
|
||||
go s.readSegmentList(stdout)
|
||||
go s.logStderr(stderr)
|
||||
go func() {
|
||||
_ = cmd.Wait()
|
||||
s.mu.Lock()
|
||||
s.ended = true
|
||||
s.mu.Unlock()
|
||||
}()
|
||||
return s, nil
|
||||
}
|
||||
|
||||
// readSegmentList 消费 ffmpeg 的 segment 完成事件流。
|
||||
func (s *stream) readSegmentList(r io.Reader) {
|
||||
scanner := bufio.NewScanner(r)
|
||||
for scanner.Scan() {
|
||||
name := scanner.Text()
|
||||
if name == "" {
|
||||
continue
|
||||
}
|
||||
path := filepath.Join(s.dir, filepath.Base(name))
|
||||
data, err := os.ReadFile(path)
|
||||
if err != nil {
|
||||
s.log.Warn("live stream read segment", zap.String("path", path), zap.Error(err))
|
||||
continue
|
||||
}
|
||||
// 诊断：TELESRV_LIVESTREAM_DUMP_DIR 非空时把原始 TS 切片留档供 ffprobe 检查。
|
||||
if dump := os.Getenv("TELESRV_LIVESTREAM_DUMP_DIR"); dump != "" {
|
||||
_ = os.MkdirAll(dump, 0o755)
|
||||
_ = os.WriteFile(filepath.Join(dump, fmt.Sprintf("ch%d_%s", s.channelID, filepath.Base(name))), data, 0o644)
|
||||
}
|
||||
_ = os.Remove(path)
|
||||
s.addSegment(data)
|
||||
}
|
||||
}
|
||||
|
||||
func (s *stream) logStderr(r io.Reader) {
|
||||
scanner := bufio.NewScanner(r)
|
||||
for scanner.Scan() {
|
||||
s.log.Info("ffmpeg", zap.Int64("channel_id", s.channelID), zap.String("line", scanner.Text()))
|
||||
}
|
||||
}
|
||||
|
||||
func (s *stream) addSegment(tsData []byte) {
|
||||
part := packUnifiedPart(tsData)
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.baseMs == 0 {
|
||||
s.baseMs = s.nowMs() / segmentDurationMs * segmentDurationMs
|
||||
}
|
||||
timeMs := s.baseMs + s.segmentSeq*segmentDurationMs
|
||||
s.segmentSeq++
|
||||
s.segments[timeMs] = part
|
||||
s.order = append(s.order, timeMs)
|
||||
s.lastMs = timeMs
|
||||
for len(s.order) > s.keep {
|
||||
delete(s.segments, s.order[0])
|
||||
s.order = s.order[1:]
|
||||
}
|
||||
s.log.Debug("live stream segment produced",
|
||||
zap.Int64("channel_id", s.channelID), zap.Int64("time_ms", timeMs),
|
||||
zap.Int("bytes", len(part)), zap.Int64("seq", s.segmentSeq),
|
||||
zap.Int64("wall_ms", s.nowMs()))
|
||||
if len(part) > 128<<10 && !s.oversizeWas {
|
||||
s.oversizeWas = true
|
||||
s.log.Warn("live stream segment exceeds 128KiB, client will truncate",
|
||||
zap.Int64("channel_id", s.channelID), zap.Int("bytes", len(part)))
|
||||
}
|
||||
}
|
||||
|
||||
// channels 返回当前时间轴（不足 minSegmentsBeforeAnnounce 段时不公布）。
|
||||
func (s *stream) channels() []domain.LiveStreamChannel {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.ended || len(s.order) < minSegmentsBeforeAnnounce {
|
||||
return nil
|
||||
}
|
||||
return []domain.LiveStreamChannel{{Channel: 1, Scale: 0, LastTimestampMs: s.lastMs}}
|
||||
}
|
||||
|
||||
// part 取指定 time_ms 的打包 part。
|
||||
func (s *stream) part(timeMs int64) ([]byte, error) {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
if s.baseMs == 0 {
|
||||
return nil, domain.ErrLiveStreamPartNotReady
|
||||
}
|
||||
if timeMs < s.baseMs || (timeMs-s.baseMs)%segmentDurationMs != 0 {
|
||||
return nil, domain.ErrLiveStreamPartExpired
|
||||
}
|
||||
if timeMs > s.lastMs {
|
||||
if s.ended {
|
||||
return nil, domain.ErrLiveStreamNoStream
|
||||
}
|
||||
return nil, domain.ErrLiveStreamPartNotReady
|
||||
}
|
||||
part, ok := s.segments[timeMs]
|
||||
if !ok {
|
||||
return nil, domain.ErrLiveStreamPartExpired
|
||||
}
|
||||
return part, nil
|
||||
}
|
||||
|
||||
func (s *stream) active() bool {
|
||||
s.mu.Lock()
|
||||
defer s.mu.Unlock()
|
||||
return !s.ended
|
||||
}
|
||||
|
||||
// stop 结束推流：断 RTMP 连接、关 ffmpeg stdin（自然退出），清空缓冲目录。
|
||||
func (s *stream) stop() {
|
||||
s.mu.Lock()
|
||||
if s.ended {
|
||||
s.mu.Unlock()
|
||||
return
|
||||
}
|
||||
s.ended = true
|
||||
s.mu.Unlock()
|
||||
if s.closer != nil {
|
||||
_ = s.closer.Close()
|
||||
}
|
||||
_ = s.stdin.Close()
|
||||
go func() {
|
||||
_ = s.cmd.Wait()
|
||||
_ = os.RemoveAll(s.dir)
|
||||
}()
|
||||
}
|
||||
|
||||
// ---- FLV 写入（RTMP tag → ffmpeg stdin）----
|
||||
|
||||
var flvHeader = []byte{'F', 'L', 'V', 0x01, 0x05, 0x00, 0x00, 0x00, 0x09, 0x00, 0x00, 0x00, 0x00}
|
||||
|
||||
// writeTag 把一条 RTMP media/data tag 以 FLV 封装写进 ffmpeg stdin。
|
||||
// tagType：8=audio 9=video 18=script data。
|
||||
func (s *stream) writeTag(tagType byte, timestampMs uint32, body []byte) error {
|
||||
s.mu.Lock()
|
||||
if s.ended {
|
||||
s.mu.Unlock()
|
||||
return domain.ErrLiveStreamNoStream
|
||||
}
|
||||
started := s.flvStarted
|
||||
s.flvStarted = true
|
||||
s.mu.Unlock()
|
||||
if !started {
|
||||
if _, err := s.stdin.Write(flvHeader); err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
var hdr [11]byte
|
||||
hdr[0] = tagType
|
||||
hdr[1] = byte(len(body) >> 16)
|
||||
hdr[2] = byte(len(body) >> 8)
|
||||
hdr[3] = byte(len(body))
|
||||
hdr[4] = byte(timestampMs >> 16)
|
||||
hdr[5] = byte(timestampMs >> 8)
|
||||
hdr[6] = byte(timestampMs)
|
||||
hdr[7] = byte(timestampMs >> 24)
|
||||
// stream id hdr[8:11] = 0
|
||||
if _, err := s.stdin.Write(hdr[:]); err != nil {
|
||||
return err
|
||||
}
|
||||
if _, err := s.stdin.Write(body); err != nil {
|
||||
return err
|
||||
}
|
||||
var prev [4]byte
|
||||
binary.BigEndian.PutUint32(prev[:], uint32(11+len(body)))
|
||||
_, err := s.stdin.Write(prev[:])
|
||||
return err
|
||||
}
|
||||
188
internal/app/livestream/service.go
Normal file
188
internal/app/livestream/service.go
Normal file
|
|
@ -0,0 +1,188 @@
|
|||
// Package livestream 实现频道 RTMP 直播的媒体面：RTMP ingest（OBS 推流）→
|
||||
// ffmpeg 转码按秒切段 → tgcalls broadcast part 内存 ring → 观众经
|
||||
// upload.getFile(inputGroupCallStream) 拉流。信令面（groupCall/participants）
|
||||
// 仍归 app/groupcalls；本包只认 stream key ↔ channelID 绑定。
|
||||
//
|
||||
// 定位：dev 主路径（单实例、内存 ring、无 CDN/多码率），
|
||||
// 生产级转码集群与分发留后续任务（见 docs/voip-module.md 直播小节）。
|
||||
package livestream
|
||||
|
||||
import (
|
||||
"context"
|
||||
"errors"
|
||||
"fmt"
|
||||
"io"
|
||||
"net"
|
||||
"os"
|
||||
"sync"
|
||||
"time"
|
||||
|
||||
rtmp "github.com/yutopp/go-rtmp"
|
||||
"go.uber.org/zap"
|
||||
|
||||
"telesrv/internal/domain"
|
||||
)
|
||||
|
||||
var errPublishRejected = errors.New("livestream: publish rejected")
|
||||
|
||||
// KeyResolver 校验 RTMP 推流密钥并返回其绑定的 channelID（app/groupcalls 实现）。
|
||||
type KeyResolver interface {
|
||||
VerifyRtmpStreamKey(ctx context.Context, key string) (channelID int64, ok bool, err error)
|
||||
}
|
||||
|
||||
// Config 是直播媒体面配置。
|
||||
type Config struct {
|
||||
// ListenAddr 是 RTMP ingest 监听地址（如 ":2400"）。
|
||||
ListenAddr string
|
||||
// FFmpegPath 是 ffmpeg 可执行文件路径（默认 "ffmpeg"，走 PATH）。
|
||||
FFmpegPath string
|
||||
// WorkDir 是切段临时目录（默认系统临时目录）。
|
||||
WorkDir string
|
||||
// SegmentKeep 是每路流内存保留的 segment 数（秒），默认 32。
|
||||
SegmentKeep int
|
||||
}
|
||||
|
||||
// Service 管理全部活跃推流会话，并向 rpc 层提供拉流查询。
|
||||
type Service struct {
|
||||
cfg Config
|
||||
keys KeyResolver
|
||||
log *zap.Logger
|
||||
|
||||
mu sync.Mutex
|
||||
streams map[int64]*stream // channelID → 活跃流
|
||||
listener net.Listener
|
||||
}
|
||||
|
||||
// NewService 创建直播服务（不监听；Start 启动 ingest）。
|
||||
func NewService(cfg Config, keys KeyResolver, log *zap.Logger) *Service {
|
||||
if cfg.FFmpegPath == "" {
|
||||
cfg.FFmpegPath = "ffmpeg"
|
||||
}
|
||||
if cfg.SegmentKeep <= 0 {
|
||||
cfg.SegmentKeep = 32
|
||||
}
|
||||
if cfg.WorkDir == "" {
|
||||
cfg.WorkDir = os.TempDir()
|
||||
}
|
||||
return &Service{cfg: cfg, keys: keys, log: log, streams: make(map[int64]*stream)}
|
||||
}
|
||||
|
||||
// Start 启动 RTMP ingest 监听。
|
||||
func (s *Service) Start() error {
|
||||
ln, err := net.Listen("tcp", s.cfg.ListenAddr)
|
||||
if err != nil {
|
||||
return fmt.Errorf("livestream: listen rtmp %s: %w", s.cfg.ListenAddr, err)
|
||||
}
|
||||
s.mu.Lock()
|
||||
s.listener = ln
|
||||
s.mu.Unlock()
|
||||
srv := rtmp.NewServer(&rtmp.ServerConfig{
|
||||
OnConnect: func(conn net.Conn) (io.ReadWriteCloser, *rtmp.ConnConfig) {
|
||||
return conn, &rtmp.ConnConfig{
|
||||
Handler: &rtmpHandler{svc: s, conn: conn},
|
||||
ControlState: rtmp.StreamControlStateConfig{
|
||||
DefaultBandwidthWindowSize: 6 * 1024 * 1024 / 8,
|
||||
},
|
||||
}
|
||||
},
|
||||
})
|
||||
go func() {
|
||||
if err := srv.Serve(ln); err != nil {
|
||||
s.log.Warn("rtmp server exited", zap.Error(err))
|
||||
}
|
||||
}()
|
||||
s.log.Info("live stream rtmp ingest listening", zap.String("addr", s.cfg.ListenAddr))
|
||||
return nil
|
||||
}
|
||||
|
||||
// Close 停止监听并结束全部推流。
|
||||
func (s *Service) Close() {
|
||||
s.mu.Lock()
|
||||
ln := s.listener
|
||||
streams := make([]*stream, 0, len(s.streams))
|
||||
for _, st := range s.streams {
|
||||
streams = append(streams, st)
|
||||
}
|
||||
s.streams = make(map[int64]*stream)
|
||||
s.mu.Unlock()
|
||||
if ln != nil {
|
||||
_ = ln.Close()
|
||||
}
|
||||
for _, st := range streams {
|
||||
st.stop()
|
||||
}
|
||||
}
|
||||
|
||||
// startPublish 鉴权 stream key 并建立一路新流；同 channel 已有活跃流时顶掉旧流
|
||||
// （OBS 断线重连的自然语义）。
|
||||
func (s *Service) startPublish(ctx context.Context, key string, conn net.Conn) (*stream, error) {
|
||||
channelID, ok, err := s.keys.VerifyRtmpStreamKey(ctx, key)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
if !ok {
|
||||
return nil, fmt.Errorf("livestream: bad stream key")
|
||||
}
|
||||
st, err := newStream(channelID, s.cfg.FFmpegPath, s.cfg.WorkDir, s.cfg.SegmentKeep, conn,
|
||||
func() int64 { return time.Now().UnixMilli() }, s.log)
|
||||
if err != nil {
|
||||
return nil, err
|
||||
}
|
||||
s.mu.Lock()
|
||||
old := s.streams[channelID]
|
||||
s.streams[channelID] = st
|
||||
s.mu.Unlock()
|
||||
if old != nil {
|
||||
old.stop()
|
||||
}
|
||||
s.log.Info("live stream publish started", zap.Int64("channel_id", channelID),
|
||||
zap.String("remote", conn.RemoteAddr().String()))
|
||||
return st, nil
|
||||
}
|
||||
|
||||
// endPublish 在推流连接断开时收尾（仅当它仍是当前流时移除）。
|
||||
func (s *Service) endPublish(st *stream) {
|
||||
s.mu.Lock()
|
||||
if s.streams[st.channelID] == st {
|
||||
delete(s.streams, st.channelID)
|
||||
}
|
||||
s.mu.Unlock()
|
||||
st.stop()
|
||||
s.log.Info("live stream publish ended", zap.Int64("channel_id", st.channelID))
|
||||
}
|
||||
|
||||
// StreamChannels 返回 channel 当前直播时间轴；无活跃推流返回空。
|
||||
func (s *Service) StreamChannels(channelID int64) []domain.LiveStreamChannel {
|
||||
s.mu.Lock()
|
||||
st := s.streams[channelID]
|
||||
s.mu.Unlock()
|
||||
if st == nil || !st.active() {
|
||||
return nil
|
||||
}
|
||||
return st.channels()
|
||||
}
|
||||
|
||||
// StreamPart 按 time_ms 取打包好的 broadcast part（仅 scale 0）。
|
||||
func (s *Service) StreamPart(channelID int64, timeMs int64, scale int) ([]byte, error) {
|
||||
if scale != 0 {
|
||||
return nil, domain.ErrLiveStreamPartExpired
|
||||
}
|
||||
s.mu.Lock()
|
||||
st := s.streams[channelID]
|
||||
s.mu.Unlock()
|
||||
if st == nil {
|
||||
return nil, domain.ErrLiveStreamNoStream
|
||||
}
|
||||
return st.part(timeMs)
|
||||
}
|
||||
|
||||
// DropChannel 断开该 channel 的推流会话并清空缓冲（discard 直播 / revoke key）。
|
||||
func (s *Service) DropChannel(channelID int64) {
|
||||
s.mu.Lock()
|
||||
st := s.streams[channelID]
|
||||
delete(s.streams, channelID)
|
||||
s.mu.Unlock()
|
||||
if st != nil {
|
||||
st.stop()
|
||||
}
|
||||
}
|
||||
Loading…
Add table
Add a link
Reference in a new issue