init
This commit is contained in:
tamakyi
@@ -0,0 +1,284 @@
|
||||
// 循环相关
|
||||
package main
|
||||
|
||||
import (
|
||||
"context"
|
||||
"time"
|
||||
)
|
||||
|
||||
// 处理管道信号
|
||||
func (s *streamer) handleMsg(msg controlMsg) {
|
||||
switch msg.c {
|
||||
case startCycle:
|
||||
lPrintln("重启监听" + s.longID() + "的直播状态")
|
||||
mainCh <- msg
|
||||
case stopCycle:
|
||||
lPrintln("退出" + s.longID() + "的循环")
|
||||
sInfoMap.Lock()
|
||||
if _, ok := sInfoMap.info[s.UID]; ok {
|
||||
delete(sInfoMap.info, s.UID)
|
||||
} else {
|
||||
lPrintErr("sInfoMap没有%s的key", s.longID())
|
||||
}
|
||||
sInfoMap.Unlock()
|
||||
case quit:
|
||||
default:
|
||||
lPrintErrf("未知的 controlMsg：%+v", msg)
|
||||
}
|
||||
}
|
||||
|
||||
// 掉线后又上线的判定窗口。AcFun 的接口和直播源偶尔会抖动一下，几秒到十几秒内
|
||||
// 「下播又上播」基本都是网络波动而不是真的换场次，窗口内恢复就当成同一场继续录，
|
||||
// 这样最后是一个成片，而不是一场直播被切成两份文件
|
||||
var liveResumeWindow = 20 * time.Second
|
||||
|
||||
// 判断主播是否还在直播（走 wap 页面）。抽成变量是为了让测试能替换掉这次网络请求
|
||||
var liveOnPage = func(s *streamer) bool { return s.isLiveOnByPage() }
|
||||
|
||||
// 断流后等待直播恢复：窗口内恢复返回 true，超时或用户主动停止返回 false
|
||||
func (s *streamer) waitResume(ctx context.Context) bool {
|
||||
deadline := time.Now().Add(liveResumeWindow)
|
||||
for {
|
||||
if ctx.Err() != nil {
|
||||
return false
|
||||
}
|
||||
// 页面说还在直播就继续；liveID 变了也接受，短暂下播又上播大多只是网络波动
|
||||
if liveOnPage(s) {
|
||||
return true
|
||||
}
|
||||
if !time.Now().Before(deadline) {
|
||||
return false
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
// 标记这条记录正在等直播恢复，这段时间里不要给同一个主播另起一场
|
||||
func markResumeWait(liveID string) {
|
||||
until := time.Now().Add(liveResumeWindow).UnixNano()
|
||||
updateLiveInfo(liveID, func(i *liveInfo) { i.resumeUntil = until })
|
||||
}
|
||||
|
||||
// 取消等待标记（恢复录制、或者本场真的结束了）
|
||||
func clearResumeWait(liveID string) {
|
||||
updateLiveInfo(liveID, func(i *liveInfo) { i.resumeUntil = 0 })
|
||||
}
|
||||
|
||||
// 主播是否还有没真正结束的录制会话：本场还有部分在录，或者断流后录制循环还在重连窗口里等。
|
||||
// cycle() 用它判断「下播又上播」是不是同一场的延续，避免同一场被重复录成两份文件
|
||||
func sessionAliveForUID(uid int) bool {
|
||||
now := time.Now().UnixNano()
|
||||
lInfoMap.RLock()
|
||||
defer lInfoMap.RUnlock()
|
||||
for _, info := range lInfoMap.info {
|
||||
if info.uid != uid {
|
||||
continue
|
||||
}
|
||||
if info.resumeUntil > now {
|
||||
return true
|
||||
}
|
||||
// 本场开始录过的部分还有没结束的，说明本场还在进行；
|
||||
// 录制循环退出时一定会标记 Done，所以这里不会把已经死掉的录制当成还活着
|
||||
if (info.recordStarted && !info.recordDone) || (info.danmuStarted && !info.danmuDone) {
|
||||
return true
|
||||
}
|
||||
}
|
||||
return false
|
||||
}
|
||||
|
||||
// 循环获取指定主播的直播状态，自动录制直播弹幕
|
||||
func (s streamer) cycle(liveID string) {
|
||||
defer func() {
|
||||
if err := recover(); err != nil {
|
||||
lPrintErr("Recovering from panic in cycle(), the error is:", err)
|
||||
lPrintErr(s.longID() + "的循环处理发生错误，尝试重启循环")
|
||||
|
||||
restart := controlMsg{s: s, c: startCycle}
|
||||
mainCh <- restart
|
||||
}
|
||||
}()
|
||||
|
||||
ch := make(chan controlMsg, 20)
|
||||
var modify bool
|
||||
sInfoMap.Lock()
|
||||
if m, ok := sInfoMap.info[s.UID]; ok {
|
||||
m.ch = ch
|
||||
modify = m.modify
|
||||
m.modify = false
|
||||
} else {
|
||||
sInfoMap.info[s.UID] = &streamerInfo{ch: ch}
|
||||
}
|
||||
sInfoMap.Unlock()
|
||||
|
||||
lPrintln("开始监听" + s.longID() + "的直播状态")
|
||||
|
||||
var isLive bool
|
||||
var lastTitle string
|
||||
var offlineAt time.Time // 首次读不到直播状态的时刻，用来判断是网络波动还是真下播
|
||||
var resumeWait bool // 已经提示过「本场还在继续」，避免每秒重复打日志
|
||||
for {
|
||||
select {
|
||||
case msg := <-ch:
|
||||
msg.liveID = liveID
|
||||
s.handleMsg(msg)
|
||||
return
|
||||
default:
|
||||
if s.isLiveOn() {
|
||||
isLive = true
|
||||
newLiveID := s.getLiveID()
|
||||
if newLiveID == "" {
|
||||
lPrintErrf("无法获取%s的liveID", s.longID())
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
|
||||
if !offlineAt.IsZero() {
|
||||
lPrintf("%s 的直播状态恢复（掉线约 %.0f 秒），判定为网络波动，继续本场录制",
|
||||
s.longID(), time.Since(offlineAt).Seconds())
|
||||
offlineAt = time.Time{}
|
||||
}
|
||||
|
||||
if liveID != newLiveID || modify {
|
||||
// 只有真正换了场次才算新开播：改设置导致的重新开始不重复推送开播通知
|
||||
newLive := liveID != newLiveID
|
||||
// 同一主播的上一场还没真正结束（断流后录制循环还在重连窗口里等），
|
||||
// 说明这次「上播」只是网络波动：先不新开一场，避免重复录制和重复通知。
|
||||
// 这里刻意不改 liveID，等本场真的结束后下一轮再按新 liveID 开一场，
|
||||
// 否则本场要是 20 秒后才结束，这一场新直播就永远开始不了了
|
||||
if newLive && !modify && sessionAliveForUID(s.UID) {
|
||||
if !resumeWait {
|
||||
resumeWait = true
|
||||
lPrintln(s.longID() + " 本场录制还在继续（判定为网络波动），不重新开始录制")
|
||||
}
|
||||
time.Sleep(time.Second)
|
||||
continue
|
||||
}
|
||||
resumeWait = false
|
||||
liveID = newLiveID
|
||||
modify = false
|
||||
title := s.getTitle()
|
||||
liveURL := s.getURL()
|
||||
lastTitle = title
|
||||
lPrintln(s.longID() + "正在直播：" + title)
|
||||
lPrintln(s.Name + "的直播观看地址：" + liveURL)
|
||||
|
||||
// 刷新主播数据，使用最新的开关设置
|
||||
if cur, ok := getStreamer(s.UID); ok {
|
||||
s = cur
|
||||
}
|
||||
if newLive && notifiedLiveOnce(liveID) {
|
||||
sendNotify(notifyEventLiveOn, s.UID, s.Name, title, liveURL, "",
|
||||
s.Name+" 开播了\n"+title+"\n"+liveURL)
|
||||
}
|
||||
filename := getTime() + " " + s.Name + " " + title
|
||||
if s.Record {
|
||||
ss := s
|
||||
safeGo(ss.longID()+"的视频录制", func() { ss.initRecord(mainCtx, liveID, filename) })
|
||||
// 没有 ffmpeg 时录不了视频，避免发出假的「开始录制」通知
|
||||
if newLive && getFFmpeg() != "" {
|
||||
sendNotify(notifyEventRecordStart, s.UID, s.Name, title, liveURL, "",
|
||||
s.Name+" 开始录制直播视频\n"+title+"\n"+liveURL)
|
||||
}
|
||||
}
|
||||
if s.Danmu {
|
||||
ss := s
|
||||
safeGo(ss.longID()+"的弹幕录制", func() { ss.initDanmu(mainCtx, liveID, filename) })
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// 应付 AcFun API 可能出现的 bug：主播没下播但 API 显示下播
|
||||
if isLive && !s.isLiveOnByPage() {
|
||||
if offlineAt.IsZero() {
|
||||
// 先不下结论：网络波动也会这样，等窗口过后再确认，免得发一条假的下播通知
|
||||
offlineAt = time.Now()
|
||||
lPrintWarn(s.longID() + " 读不到直播状态，等待 " + itoa(int(liveResumeWindow.Seconds())) +
|
||||
" 秒确认（网络波动会继续本场录制）")
|
||||
} else if time.Since(offlineAt) >= liveResumeWindow {
|
||||
isLive = false
|
||||
offlineAt = time.Time{}
|
||||
resumeWait = false
|
||||
lPrintln(s.longID() + "已经下播")
|
||||
sendNotify(notifyEventLiveOff, s.UID, s.Name, lastTitle, s.getURL(), "",
|
||||
s.Name+" 已经下播\n最后一场："+lastTitle)
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
time.Sleep(time.Second)
|
||||
}
|
||||
}
|
||||
|
||||
// 录制全部结束、且没有再产生合并任务后，liveInfo 的保留时长
|
||||
const liveInfoKeepDuration = 30 * time.Minute
|
||||
|
||||
// 循环检测删除 lInfoMap.info 里没有录制弹幕的 key
|
||||
func cycleDelKey(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
sweepLiveInfo()
|
||||
|
||||
// 每分钟循环一次
|
||||
time.Sleep(time.Minute)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// 清理已经彻底结束的直播记录，返回删除的条数
|
||||
func sweepLiveInfo() int {
|
||||
now := time.Now()
|
||||
nowNs := now.UnixNano()
|
||||
removed := 0
|
||||
lInfoMap.Lock()
|
||||
defer lInfoMap.Unlock()
|
||||
for liveID, info := range lInfoMap.info {
|
||||
// 正在录制或正在合并的记录不能删
|
||||
if info.isDanmu || info.isRecording || info.mergeStarted {
|
||||
continue
|
||||
}
|
||||
// 断流后录制循环还在重连窗口里等着，这条记录后面还要接着用
|
||||
if info.resumeUntil > nowNs {
|
||||
continue
|
||||
}
|
||||
// 刚结束录制的记录要留住：弹幕和录播的结束时间可能相差较久（录播分段之间还有间隔），
|
||||
// 删掉 key 会让后结束的那部分在 finishPart 里找不到记录，自动合并被静默跳过
|
||||
if info.danmuDone || info.recordDone {
|
||||
if info.doneAt == 0 || now.Sub(time.Unix(0, info.doneAt)) < liveInfoKeepDuration {
|
||||
continue
|
||||
}
|
||||
}
|
||||
delete(lInfoMap.info, liveID)
|
||||
removed++
|
||||
}
|
||||
return removed
|
||||
}
|
||||
|
||||
// 循环获取 AcFun 直播间数据
|
||||
func cycleFetch(ctx context.Context) {
|
||||
for {
|
||||
select {
|
||||
case <-ctx.Done():
|
||||
return
|
||||
default:
|
||||
if ok := fetchAllRooms(); ok {
|
||||
if len(liveRooms.newRooms) == 0 {
|
||||
lPrintWarn("没有人在直播")
|
||||
}
|
||||
|
||||
liveRooms.Lock()
|
||||
for uid, room := range liveRooms.rooms {
|
||||
delete(liveRooms.rooms, uid)
|
||||
liveRoomPool.Put(room)
|
||||
}
|
||||
liveRooms.rooms = liveRooms.newRooms
|
||||
liveRooms.Unlock()
|
||||
}
|
||||
|
||||
// 每 10 秒循环一次
|
||||
time.Sleep(10 * time.Second)
|
||||
}
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user