From 37d4b6d14b011f4ded09079325791209f652bb61 Mon Sep 17 00:00:00 2001 From: joestarzxh Date: Thu, 23 Apr 2026 11:02:15 +0800 Subject: [PATCH] =?UTF-8?q?[fix]=E8=B0=83=E6=95=B4gb28181=E7=9B=AE?= =?UTF-8?q?=E5=BD=95=E7=BB=93=E6=9E=84=E7=AD=89?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- gb28181/mediaserver/conn.go | 34 ++- gb28181/mediaserver/server.go | 56 +++- gb28181/rtppub/manager.go | 280 ++++++++++++++++++ gb28181/rtppub/manager_test.go | 127 ++++++++ .../lower_push_session.go | 2 +- .../lower_push_session_test.go | 2 +- gb28181/server.go | 11 +- logic/group.go | 61 ++-- logic/group_manager.go | 73 ++++- logic/group_test.go | 202 ++++++++++++- server/logs/lalserver.log | 210 +++++++++++++ server/router.go | 55 ++-- server/router_test.go | 48 ++- server/server.go | 23 +- 14 files changed, 1083 insertions(+), 101 deletions(-) create mode 100644 gb28181/rtppub/manager.go create mode 100644 gb28181/rtppub/manager_test.go rename gb28181/{mediaserver => rtppush}/lower_push_session.go (99%) rename gb28181/{mediaserver => rtppush}/lower_push_session_test.go (99%) create mode 100644 server/logs/lalserver.log diff --git a/gb28181/mediaserver/conn.go b/gb28181/mediaserver/conn.go index 4720d69..c074b9a 100644 --- a/gb28181/mediaserver/conn.go +++ b/gb28181/mediaserver/conn.go @@ -51,9 +51,11 @@ type Conn struct { buffer *bytes.Buffer key string - mediaServer *GB28181MediaServer - one sync.Once - oneSaveConn sync.Once + mediaServer *GB28181MediaServer + preferMediaKeyLookup bool + readTimeout time.Duration + one sync.Once + oneSaveConn sync.Once } func NewConn(conn net.Conn, observer IGbObserver, lal logic.ILalServer) *Conn { @@ -76,12 +78,18 @@ func (c *Conn) SetMediaServer(mediaServer *GB28181MediaServer) { func (c *Conn) SetKey(key string) { c.key = key } +func (c *Conn) SetPreferMediaKeyLookup(prefer bool) { + c.preferMediaKeyLookup = prefer +} +func (c *Conn) SetReadTimeout(timeout time.Duration) { + c.readTimeout = timeout +} func (c *Conn) Serve() (err error) { defer func() { nazalog.Info("conn close, err:", err) c.Close() - if c.observer != nil { + if c.observer != nil && c.streamName != "" { c.observer.NotifyClose(c.streamName) } if c.psDumpFile != nil { @@ -95,7 +103,9 @@ func (c *Conn) Serve() (err error) { nazalog.Info("gb28181 conn, remoteaddr:", c.conn.RemoteAddr().String(), " localaddr:", c.conn.LocalAddr().String()) for { - c.conn.SetReadDeadline(time.Now().Add(10 * time.Second)) + if c.readTimeout > 0 { + c.conn.SetReadDeadline(time.Now().Add(c.readTimeout)) + } pkt := &rtp.Packet{} if c.conn.RemoteAddr().Network() == "udp" { buf := make([]byte, 1472*4) @@ -132,7 +142,13 @@ func (c *Conn) Serve() (err error) { if !c.check && c.observer != nil { var mediaInfo *MediaInfo var ok bool - if pkt.SSRC != 0 { + if c.preferMediaKeyLookup { + mediaInfo, ok = c.observer.GetMediaInfoByKey(c.key) + if !ok { + nazalog.Error("get mediaInfo :", c.key) + return fmt.Errorf("get mediaInfo:%s", c.key) + } + } else if pkt.SSRC != 0 { mediaInfo, ok = c.observer.CheckSsrc(pkt.SSRC) if !ok { nazalog.Error("invalid ssrc:", pkt.SSRC) @@ -159,6 +175,9 @@ func (c *Conn) Serve() (err error) { } } nazalog.Info("gb28181 ssrc check success, streamName:", c.streamName) + if c.observer != nil { + c.observer.OnRtpPacket(c.streamName, c.key) + } session, err := c.lalServer.AddCustomizePubSession(mediaInfo.StreamName) if err != nil { @@ -174,6 +193,9 @@ func (c *Conn) Serve() (err error) { c.lalSession = session } c.rtpPts = uint64(pkt.Header.Timestamp) + if c.observer != nil && c.streamName != "" { + c.observer.OnRtpPacket(c.streamName, c.key) + } if c.demuxer != nil { if c.psDumpFile != nil { c.psDumpFile.WriteWithType(pkt.Payload, base.DumpTypePsRtpData) diff --git a/gb28181/mediaserver/server.go b/gb28181/mediaserver/server.go index c40258c..4838650 100644 --- a/gb28181/mediaserver/server.go +++ b/gb28181/mediaserver/server.go @@ -4,16 +4,20 @@ import ( "errors" "net" "sync" + "sync/atomic" "time" "github.com/q191201771/lal/pkg/logic" "github.com/q191201771/naza/pkg/nazalog" ) +const defaultReadTimeout = 10 * time.Second + type IGbObserver interface { CheckSsrc(ssrc uint32) (*MediaInfo, bool) GetMediaInfoByKey(key string) (*MediaInfo, bool) NotifyClose(streamName string) + OnRtpPacket(streamName string, mediaKey string) } type GB28181MediaServer struct { @@ -22,52 +26,79 @@ type GB28181MediaServer struct { listener net.Listener - disposeOnce sync.Once - observer IGbObserver - mediaKey string + disposeOnce sync.Once + disposed atomic.Bool + observer IGbObserver + mediaKey string + preferMediaKeyLookup bool + readTimeout time.Duration conns sync.Map //增加链接对象，目前只适用于多端口 } func NewGB28181MediaServer(listenPort int, mediaKey string, observer IGbObserver, lal logic.ILalServer) *GB28181MediaServer { return &GB28181MediaServer{ - listenPort: listenPort, - lalServer: lal, - observer: observer, - mediaKey: mediaKey, + listenPort: listenPort, + lalServer: lal, + observer: observer, + mediaKey: mediaKey, + readTimeout: defaultReadTimeout, } } + +func (s *GB28181MediaServer) WithPreferMediaKeyLookup(prefer bool) *GB28181MediaServer { + s.preferMediaKeyLookup = prefer + return s +} + +func (s *GB28181MediaServer) WithReadTimeout(timeout time.Duration) *GB28181MediaServer { + s.readTimeout = timeout + return s +} + func (s *GB28181MediaServer) GetListenerPort() uint16 { return uint16(s.listenPort) } func (s *GB28181MediaServer) Start(listener net.Listener) (err error) { s.listener = listener - if s.listener != nil { - go func() { + if listener != nil { + go func(listener net.Listener) { for { - if s.listener == nil { + if s.disposed.Load() { return } - conn, err := s.listener.Accept() + conn, err := listener.Accept() if err != nil { var ne net.Error if ok := errors.As(err, &ne); ok && ne.Timeout() { nazalog.Error("Accept failed: timeout error, retrying...") time.Sleep(time.Second / 20) + continue } else { break } } + if conn == nil { + continue + } + if s.disposed.Load() { + conn.Close() + return + } c := NewConn(conn, s.observer, s.lalServer) c.SetKey(s.mediaKey) c.SetMediaServer(s) + c.SetPreferMediaKeyLookup(s.preferMediaKeyLookup) + c.SetReadTimeout(s.readTimeout) + s.conns.Store(c, c) go func() { c.Serve() + s.conns.Delete(c) s.conns.Delete(c.streamName) }() } - }() + }(listener) } return } @@ -79,6 +110,7 @@ func (s *GB28181MediaServer) CloseConn(streamName string) { } func (s *GB28181MediaServer) Dispose() { s.disposeOnce.Do(func() { + s.disposed.Store(true) s.conns.Range(func(_, value any) bool { conn := value.(*Conn) conn.Close() diff --git a/gb28181/rtppub/manager.go b/gb28181/rtppub/manager.go new file mode 100644 index 0000000..dcbdc4a --- /dev/null +++ b/gb28181/rtppub/manager.go @@ -0,0 +1,280 @@ +package rtppub + +import ( + "errors" + "fmt" + "net" + "sync" + "time" + + udpTransport "github.com/pion/transport/v3/udp" + "github.com/q191201771/lal/pkg/base" + "github.com/q191201771/lal/pkg/logic" + "github.com/q191201771/lalmax/config" + "github.com/q191201771/lalmax/gb28181/mediaserver" + "github.com/q191201771/naza/pkg/nazalog" +) + +const ( + defaultPortMin = 30000 + defaultPortMaxIncrement = 3000 +) + +var ( + errDuplicateStream = errors.New("rtp pub stream already exists") + errSessionNotFound = errors.New("rtp pub session not found") +) + +type Manager struct { + mu sync.Mutex + + lalServer logic.ILalServer + portMin int + portMax int + + sessionsByID map[string]*Session + sessionsByStream map[string]*Session + sessionsByKey map[string]*Session +} + +type Session struct { + ID string + StreamName string + MediaKey string + Network string + Port int + + mediaInfo mediaserver.MediaInfo + server *mediaserver.GB28181MediaServer + lastActive time.Time + done chan struct{} + closeOnce sync.Once +} + +func NewManager(lalServer logic.ILalServer, mediaConfig config.GB28181MediaConfig) *Manager { + basePort := int(mediaConfig.ListenPort) + if basePort == 0 { + basePort = defaultPortMin + } + maxIncrement := mediaConfig.MultiPortMaxIncrement + if maxIncrement == 0 { + maxIncrement = defaultPortMaxIncrement + } + + portMin := basePort + if mediaConfig.ListenPort != 0 { + portMin++ + } + + return &Manager{ + lalServer: lalServer, + portMin: portMin, + portMax: basePort + int(maxIncrement), + sessionsByID: make(map[string]*Session), + sessionsByStream: make(map[string]*Session), + sessionsByKey: make(map[string]*Session), + } +} + +func (m *Manager) Start(req base.ApiCtrlStartRtpPubReq) (ret base.ApiCtrlStartRtpPubResp) { + if req.StreamName == "" { + ret.ErrorCode = base.ErrorCodeParamMissing + ret.Desp = base.DespParamMissing + return + } + + network := "udp" + if req.IsTcpFlag != 0 { + network = "tcp" + } + + m.mu.Lock() + if _, ok := m.sessionsByStream[req.StreamName]; ok { + m.mu.Unlock() + ret.ErrorCode = base.ErrorCodeListenUdpPortFail + ret.Desp = errDuplicateStream.Error() + return + } + m.mu.Unlock() + + listener, port, err := m.listen(req.Port, network) + if err != nil { + ret.ErrorCode = base.ErrorCodeListenUdpPortFail + ret.Desp = err.Error() + return + } + + sessionID := base.GenUkPsPubSession() + mediaKey := fmt.Sprintf("%s%d", network, port) + session := &Session{ + ID: sessionID, + StreamName: req.StreamName, + MediaKey: mediaKey, + Network: network, + Port: port, + mediaInfo: mediaserver.MediaInfo{ + StreamName: req.StreamName, + DumpFileName: req.DebugDumpPacket, + MediaKey: mediaKey, + }, + lastActive: time.Now(), + done: make(chan struct{}), + } + readTimeout := time.Duration(req.TimeoutMs) * time.Millisecond + session.server = mediaserver.NewGB28181MediaServer(port, mediaKey, m, m.lalServer). + WithPreferMediaKeyLookup(true). + WithReadTimeout(readTimeout) + + m.mu.Lock() + if _, ok := m.sessionsByStream[req.StreamName]; ok { + m.mu.Unlock() + _ = listener.Close() + ret.ErrorCode = base.ErrorCodeListenUdpPortFail + ret.Desp = errDuplicateStream.Error() + return + } + m.sessionsByID[session.ID] = session + m.sessionsByStream[session.StreamName] = session + m.sessionsByKey[session.MediaKey] = session + m.mu.Unlock() + + if err = session.server.Start(listener); err != nil { + m.stopSession(session) + ret.ErrorCode = base.ErrorCodeListenUdpPortFail + ret.Desp = err.Error() + return + } + + if req.TimeoutMs > 0 { + go m.watchTimeout(session, time.Duration(req.TimeoutMs)*time.Millisecond) + } + + ret.ErrorCode = base.ErrorCodeSucc + ret.Desp = base.DespSucc + ret.Data.SessionId = session.ID + ret.Data.StreamName = session.StreamName + ret.Data.Port = session.Port + return +} + +func (m *Manager) Stop(streamName, sessionID string) (*Session, error) { + m.mu.Lock() + var session *Session + if sessionID != "" { + session = m.sessionsByID[sessionID] + } else if streamName != "" { + session = m.sessionsByStream[streamName] + } + m.mu.Unlock() + + if session == nil { + return nil, errSessionNotFound + } + + m.stopSession(session) + return session, nil +} + +func (m *Manager) GetMediaInfoByKey(key string) (*mediaserver.MediaInfo, bool) { + m.mu.Lock() + defer m.mu.Unlock() + + session, ok := m.sessionsByKey[key] + if !ok { + return nil, false + } + return &session.mediaInfo, true +} + +func (m *Manager) CheckSsrc(ssrc uint32) (*mediaserver.MediaInfo, bool) { + return nil, false +} + +func (m *Manager) NotifyClose(streamName string) { +} + +func (m *Manager) OnRtpPacket(streamName string, mediaKey string) { + m.mu.Lock() + defer m.mu.Unlock() + + if session, ok := m.sessionsByKey[mediaKey]; ok { + session.lastActive = time.Now() + } +} + +func (m *Manager) stopSession(session *Session) { + m.mu.Lock() + if current := m.sessionsByID[session.ID]; current != session { + m.mu.Unlock() + return + } + delete(m.sessionsByID, session.ID) + delete(m.sessionsByStream, session.StreamName) + delete(m.sessionsByKey, session.MediaKey) + session.closeOnce.Do(func() { + close(session.done) + }) + m.mu.Unlock() + + session.server.Dispose() +} + +func (m *Manager) watchTimeout(session *Session, timeout time.Duration) { + interval := timeout / 2 + if interval < time.Second { + interval = time.Second + } + + ticker := time.NewTicker(interval) + defer ticker.Stop() + + for { + select { + case <-session.done: + return + case <-ticker.C: + m.mu.Lock() + current := m.sessionsByID[session.ID] + expired := current == session && time.Since(session.lastActive) >= timeout + m.mu.Unlock() + if expired { + nazalog.Warnf("rtp pub timeout, streamName:%s, sessionId:%s", session.StreamName, session.ID) + m.stopSession(session) + return + } + } + } +} + +func (m *Manager) listen(port int, network string) (net.Listener, int, error) { + if port > 0 { + listener, err := listenPort(port, network) + return listener, port, err + } + + var lastErr error + for p := m.portMin; p <= m.portMax; p++ { + listener, err := listenPort(p, network) + if err == nil { + return listener, p, nil + } + lastErr = err + } + if lastErr == nil { + lastErr = fmt.Errorf("no available %s port in range [%d,%d]", network, m.portMin, m.portMax) + } + return nil, 0, lastErr +} + +func listenPort(port int, network string) (net.Listener, error) { + addr := fmt.Sprintf(":%d", port) + if network == "tcp" { + return net.Listen("tcp", addr) + } + + udpAddr, err := net.ResolveUDPAddr("udp", addr) + if err != nil { + return nil, err + } + return udpTransport.Listen("udp", udpAddr) +} diff --git a/gb28181/rtppub/manager_test.go b/gb28181/rtppub/manager_test.go new file mode 100644 index 0000000..4fb2aaf --- /dev/null +++ b/gb28181/rtppub/manager_test.go @@ -0,0 +1,127 @@ +package rtppub + +import ( + "errors" + "net" + "testing" + "time" + + "github.com/q191201771/lal/pkg/base" + "github.com/q191201771/lalmax/config" +) + +func freeTCPPort(t *testing.T) uint16 { + t.Helper() + + listener, err := net.Listen("tcp", "127.0.0.1:0") + if err != nil { + t.Fatal(err) + } + defer listener.Close() + + return uint16(listener.Addr().(*net.TCPAddr).Port) +} + +func newTestManager(t *testing.T) *Manager { + t.Helper() + + return NewManager(nil, config.GB28181MediaConfig{}) +} + +func TestManagerStartStopBySessionID(t *testing.T) { + manager := newTestManager(t) + + resp := manager.Start(base.ApiCtrlStartRtpPubReq{ + StreamName: "rtp-pub-start-stop", + Port: int(freeTCPPort(t)), + TimeoutMs: 0, + IsTcpFlag: 1, + }) + if resp.ErrorCode != base.ErrorCodeSucc { + t.Fatalf("start failed, code=%d desp=%s", resp.ErrorCode, resp.Desp) + } + if resp.Data.SessionId == "" || resp.Data.Port == 0 { + t.Fatalf("unexpected start response: %+v", resp.Data) + } + + session, err := manager.Stop("", resp.Data.SessionId) + if err != nil { + t.Fatal(err) + } + if session.ID != resp.Data.SessionId { + t.Fatalf("stopped session id = %s, want %s", session.ID, resp.Data.SessionId) + } + + if _, err = manager.Stop("", resp.Data.SessionId); !errors.Is(err, errSessionNotFound) { + t.Fatalf("second stop err = %v, want %v", err, errSessionNotFound) + } +} + +func TestManagerRejectsDuplicateStream(t *testing.T) { + manager := newTestManager(t) + + resp := manager.Start(base.ApiCtrlStartRtpPubReq{ + StreamName: "rtp-pub-duplicate", + Port: int(freeTCPPort(t)), + TimeoutMs: 0, + IsTcpFlag: 1, + }) + if resp.ErrorCode != base.ErrorCodeSucc { + t.Fatalf("start failed, code=%d desp=%s", resp.ErrorCode, resp.Desp) + } + defer manager.Stop(resp.Data.StreamName, "") + + duplicate := manager.Start(base.ApiCtrlStartRtpPubReq{ + StreamName: "rtp-pub-duplicate", + Port: int(freeTCPPort(t)), + TimeoutMs: 0, + IsTcpFlag: 1, + }) + if duplicate.ErrorCode == base.ErrorCodeSucc { + t.Fatalf("duplicate stream start unexpectedly succeeded: %+v", duplicate.Data) + } +} + +func TestManagerTimeoutRemovesIdleSession(t *testing.T) { + manager := newTestManager(t) + + resp := manager.Start(base.ApiCtrlStartRtpPubReq{ + StreamName: "rtp-pub-timeout", + Port: int(freeTCPPort(t)), + TimeoutMs: 10, + IsTcpFlag: 1, + }) + if resp.ErrorCode != base.ErrorCodeSucc { + t.Fatalf("start failed, code=%d desp=%s", resp.ErrorCode, resp.Desp) + } + defer manager.Stop(resp.Data.StreamName, "") + + deadline := time.After(2 * time.Second) + ticker := time.NewTicker(20 * time.Millisecond) + defer ticker.Stop() + + for { + select { + case <-deadline: + t.Fatal("session was not removed after timeout") + case <-ticker.C: + manager.mu.Lock() + _, ok := manager.sessionsByID[resp.Data.SessionId] + manager.mu.Unlock() + if !ok { + return + } + } + } +} + +func TestNewManagerUsesConfiguredPortRangeAfterListenPort(t *testing.T) { + manager := NewManager(nil, config.GB28181MediaConfig{ + ListenPort: 31000, + MultiPortMaxIncrement: 10, + }) + + if manager.portMin != 31001 || manager.portMax != 31010 { + t.Fatalf("port range = [%d,%d], want [31001,31010]", manager.portMin, manager.portMax) + } +} diff --git a/gb28181/mediaserver/lower_push_session.go b/gb28181/rtppush/lower_push_session.go similarity index 99% rename from gb28181/mediaserver/lower_push_session.go rename to gb28181/rtppush/lower_push_session.go index 740d490..0b6c619 100644 --- a/gb28181/mediaserver/lower_push_session.go +++ b/gb28181/rtppush/lower_push_session.go @@ -1,4 +1,4 @@ -package mediaserver +package rtppush import ( "fmt" diff --git a/gb28181/mediaserver/lower_push_session_test.go b/gb28181/rtppush/lower_push_session_test.go similarity index 99% rename from gb28181/mediaserver/lower_push_session_test.go rename to gb28181/rtppush/lower_push_session_test.go index 62b57c9..c87aea9 100644 --- a/gb28181/mediaserver/lower_push_session_test.go +++ b/gb28181/rtppush/lower_push_session_test.go @@ -1,4 +1,4 @@ -package mediaserver +package rtppush import ( "io" diff --git a/gb28181/server.go b/gb28181/server.go index 72ac3db..c85ada9 100644 --- a/gb28181/server.go +++ b/gb28181/server.go @@ -182,7 +182,7 @@ func (s *GB28181Server) OnStartMediaServer(netWork string, singlePort bool, devi mediasvr = mediaserver.NewGB28181MediaServer(int(s.conf.MediaConfig.ListenPort), fmt.Sprintf("%s%d", "tcp", s.conf.MediaConfig.ListenPort), s, s.lalServer) listener, err = s.tcpAvailConnPool.ListenWithPort(s.conf.MediaConfig.ListenPort) if err != nil { - nazalog.Error("gb28181 media server tcp Listen failed:%s", err.Error()) + nazalog.Errorf("gb28181 media server tcp Listen failed:%s", err.Error()) return nil } s.MediaServerMap.Store(fmt.Sprintf("%s%d", "tcp", s.conf.MediaConfig.ListenPort), mediasvr) @@ -190,7 +190,7 @@ func (s *GB28181Server) OnStartMediaServer(netWork string, singlePort bool, devi mediasvr = mediaserver.NewGB28181MediaServer(int(s.conf.MediaConfig.ListenPort), fmt.Sprintf("%s%d", "udp", s.conf.MediaConfig.ListenPort), s, s.lalServer) listener, err = s.udpAvailConnPool.ListenWithPort(s.conf.MediaConfig.ListenPort) if err != nil { - nazalog.Error("gb28181 media server udp Listen failed:%s", err.Error()) + nazalog.Errorf("gb28181 media server udp Listen failed:%s", err.Error()) return nil } s.MediaServerMap.Store(fmt.Sprintf("%s%d", "udp", s.conf.MediaConfig.ListenPort), mediasvr) @@ -200,14 +200,14 @@ func (s *GB28181Server) OnStartMediaServer(netWork string, singlePort bool, devi if isTcpFlag { listener, port, err = s.tcpAvailConnPool.Acquire() if err != nil { - nazalog.Error("gb28181 media server tcp acquire failed:%s", err.Error()) + nazalog.Errorf("gb28181 media server tcp acquire failed:%s", err.Error()) return nil } mediaKey = fmt.Sprintf("%s%d", "tcp", port) } else { listener, port, err = s.udpAvailConnPool.Acquire() if err != nil { - nazalog.Error("gb28181 media server udp acquire failed:%s", err.Error()) + nazalog.Errorf("gb28181 media server udp acquire failed:%s", err.Error()) return nil } mediaKey = fmt.Sprintf("%s%d", "udp", port) @@ -336,6 +336,9 @@ func (s *GB28181Server) NotifyClose(streamName string) { }) } +func (s *GB28181Server) OnRtpPacket(streamName string, mediaKey string) { +} + func (s *GB28181Server) startJob() { statusTick := time.NewTicker(s.HeartbeatInterval / 2) banTick := time.NewTicker(s.RemoveBanInterval) diff --git a/logic/group.go b/logic/group.go index 38422e4..941dd4c 100644 --- a/logic/group.go +++ b/logic/group.go @@ -44,13 +44,14 @@ type Group struct { key StreamKey consumers sync.Map hlssvr *hls.HlsServer + manager *ComplexGroupManager gopCache *GopCache gopCacheMux sync.RWMutex lifecycleMux sync.RWMutex stopOnce sync.Once msgMux sync.Mutex hasVideo bool - closed bool + closed atomic.Bool } type subscriberState struct { @@ -95,26 +96,33 @@ func (s *subscriberState) Url() string { return s.key.String() } -func NewGroup(uniqueKey string, key StreamKey, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *Group { +func newGroup(manager *ComplexGroupManager, uniqueKey string, key StreamKey, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *Group { group := &Group{ uniqueKey: uniqueKey, key: key, hlssvr: hlssvr, + manager: manager, gopCache: NewGopCache(gopNum, singleGopMaxFrameNum), } - if group.hlssvr != nil { - group.hlssvr.NewHlsSessionWithAppName(key.AppName, key.StreamName) - } - nazalog.Infof("create group, uniqueKey:%s, streamKey:%s", uniqueKey, key.String()) - GetGroupManagerInstance().SetGroup(key, group) return group } -func NewGroupByStreamName(uniqueKey, streamName string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) *Group { - return NewGroup(uniqueKey, StreamKeyFromStreamName(streamName), hlssvr, gopNum, singleGopMaxFrameNum) +func (group *Group) initHlsSession() { + if group != nil && group.hlssvr != nil { + group.hlssvr.NewHlsSessionWithAppName(group.key.AppName, group.key.StreamName) + } +} + +func (group *Group) waitLifecycleIdle() { + if group == nil { + return + } + + group.lifecycleMux.RLock() + group.lifecycleMux.RUnlock() } func (group *Group) Key() StreamKey { @@ -127,7 +135,7 @@ func (group *Group) UniqueKey() string { func (group *Group) OnMsg(msg base.RtmpMsg) { group.lifecycleMux.RLock() - if group.closed { + if group.closed.Load() { group.lifecycleMux.RUnlock() return } @@ -164,7 +172,7 @@ func (group *Group) OnMsg(msg base.RtmpMsg) { func (group *Group) OnStop() { group.stopOnce.Do(func() { group.lifecycleMux.Lock() - group.closed = true + group.closed.Store(true) if group.hlssvr != nil { group.hlssvr.OnStopWithAppName(group.key.AppName, group.key.StreamName) @@ -186,7 +194,9 @@ func (group *Group) OnStop() { c.stopWithNotify() } - GetGroupManagerInstance().RemoveGroupIfMatch(group.key, group) + if group.manager != nil { + group.manager.RemoveGroupIfMatch(group.key, group) + } }) } @@ -204,7 +214,7 @@ func (group *Group) AddSubscriberWithReplay(info SubscriberInfo, subscriber Subs } group.lifecycleMux.RLock() - if group.closed { + if group.closed.Load() { group.lifecycleMux.RUnlock() nazalog.Warnf("AddSubscriber skipped, group is closed, streamKey:%s, subscriberId:%s", group.key.String(), info.SubscriberID) return @@ -328,18 +338,22 @@ func (group *Group) handleSubscriberMsg(c *subscriberState, msg base.RtmpMsg, ha return } if v := group.GetVideoSeqHeaderMsg(); v != nil { - c.subscriber.OnMsg(*v) + if !c.deliverMsg(*v) { + return + } } if v := group.GetAudioSeqHeaderMsg(); v != nil && v.IsAacSeqHeader() { - c.subscriber.OnMsg(*v) + if !c.deliverMsg(*v) { + return + } } c.hasSendVideo = true } - c.subscriber.OnMsg(msg) + c.deliverMsg(msg) } else if msg.Header.MsgTypeId == base.RtmpTypeIdAudio { if !hasVideo || c.hasSendVideo { - c.subscriber.OnMsg(msg) + c.deliverMsg(msg) } } } @@ -359,11 +373,22 @@ func (group *Group) replayGopMessagesLocked(c *subscriberState, msgs []base.Rtmp } for _, msg := range msgs { - c.subscriber.OnMsg(msg) + if !c.deliverMsg(msg) { + return + } } c.hasSendVideo = true } +func (s *subscriberState) deliverMsg(msg base.RtmpMsg) bool { + if s == nil || s.stopped.Load() || s.subscriber == nil { + return false + } + + s.subscriber.OnMsg(msg) + return !s.stopped.Load() && s.subscriber != nil +} + func (s *subscriberState) stopWithNotify() { if s == nil { return diff --git a/logic/group_manager.go b/logic/group_manager.go index 2212105..2b03988 100644 --- a/logic/group_manager.go +++ b/logic/group_manager.go @@ -3,11 +3,12 @@ package logic import ( "sync" + "github.com/q191201771/lalmax/fmp4/hls" "github.com/q191201771/naza/pkg/nazalog" ) type IGroupManager interface { - SetGroup(key StreamKey, group *Group) + GetOrCreateGroup(key StreamKey, uniqueKey string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) (*Group, bool) RemoveGroup(key StreamKey) RemoveGroupIfMatch(key StreamKey, group *Group) GetGroup(key StreamKey) (bool, *Group) @@ -42,16 +43,68 @@ func GetGroupManagerInstance() *ComplexGroupManager { return defaultGroupManager } -func (m *ComplexGroupManager) SetGroup(key StreamKey, group *Group) { +func (m *ComplexGroupManager) GetOrCreateGroup(key StreamKey, uniqueKey string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) (*Group, bool) { + if m == nil || !key.Valid() { + return nil, false + } + + for { + m.mutex.Lock() + ok, existing := m.getGroupLocked(key) + if !ok { + break + } + if !existing.closed.Load() { + m.mutex.Unlock() + return existing, false + } + m.mutex.Unlock() + + // 等旧 group 完成 HLS 清理后再发布替换 group， + // 否则旧 group 的 OnStop 可能删掉新的 HLS session。 + existing.waitLifecycleIdle() + + m.mutex.Lock() + ok, current := m.getGroupLocked(key) + if !ok { + break + } + if current == existing { + break + } + if !current.closed.Load() { + m.mutex.Unlock() + return current, false + } + m.mutex.Unlock() + } + + group := newGroup(m, uniqueKey, key, hlssvr, gopNum, singleGopMaxFrameNum) + group.initHlsSession() + m.setGroupLocked(key, group) + m.mutex.Unlock() + return group, true +} + +func (m *ComplexGroupManager) GetOrCreateGroupByStreamName(uniqueKey, streamName string, hlssvr *hls.HlsServer, gopNum, singleGopMaxFrameNum int) (*Group, bool) { + return m.GetOrCreateGroup(StreamKeyFromStreamName(streamName), uniqueKey, hlssvr, gopNum, singleGopMaxFrameNum) +} + +func (m *ComplexGroupManager) setGroup(key StreamKey, group *Group) { if m == nil || !key.Valid() || group == nil { return } - nazalog.Info("SetGroup, streamKey:", key.String()) - m.mutex.Lock() defer m.mutex.Unlock() + m.setGroupLocked(key, group) +} + +func (m *ComplexGroupManager) setGroupLocked(key StreamKey, group *Group) { + nazalog.Info("SetGroup, streamKey:", key.String()) + + group.manager = m if key.AppName == "" { m.onlyStreamNameGroups[key.StreamName] = group return @@ -65,8 +118,8 @@ func (m *ComplexGroupManager) SetGroup(key StreamKey, group *Group) { groups[key.StreamName] = group } -func (m *ComplexGroupManager) SetGroupByStreamName(streamName string, group *Group) { - m.SetGroup(StreamKeyFromStreamName(streamName), group) +func (m *ComplexGroupManager) setGroupByStreamName(streamName string, group *Group) { + m.setGroup(StreamKeyFromStreamName(streamName), group) } func (m *ComplexGroupManager) RemoveGroup(key StreamKey) { @@ -130,11 +183,15 @@ func (m *ComplexGroupManager) GetGroup(key StreamKey) (bool, *Group) { m.mutex.RLock() defer m.mutex.RUnlock() + return m.getGroupLocked(key) +} + +func (m *ComplexGroupManager) getGroupLocked(key StreamKey) (bool, *Group) { if key.AppName == "" { if group, ok := m.onlyStreamNameGroups[key.StreamName]; ok { return true, group } - return m.getGroupByOnlyStreamName(key.StreamName) + return m.getGroupByOnlyStreamNameLocked(key.StreamName) } if groups, ok := m.appNameStreamNameGroups[key.AppName]; ok { @@ -155,7 +212,7 @@ func (m *ComplexGroupManager) GetGroupByStreamName(streamName string) (bool, *Gr } // streamName 单独查找只在匹配唯一 appName 时成功，避免跨 app 串流。 -func (m *ComplexGroupManager) getGroupByOnlyStreamName(streamName string) (bool, *Group) { +func (m *ComplexGroupManager) getGroupByOnlyStreamNameLocked(streamName string) (bool, *Group) { var found *Group matchCount := 0 for _, groups := range m.appNameStreamNameGroups { diff --git a/logic/group_test.go b/logic/group_test.go index b8e70fd..47fddd4 100644 --- a/logic/group_test.go +++ b/logic/group_test.go @@ -99,6 +99,39 @@ func (s *blockingSubscriber) markers() []byte { return out } +type selfRemovingSubscriber struct { + group *Group + id string + + mu sync.Mutex + msgs []base.RtmpMsg +} + +func (s *selfRemovingSubscriber) OnMsg(msg base.RtmpMsg) { + s.mu.Lock() + s.msgs = append(s.msgs, msg.Clone()) + shouldRemove := len(s.msgs) == 1 + s.mu.Unlock() + + if shouldRemove { + s.group.RemoveConsumer(s.id) + } +} + +func (s *selfRemovingSubscriber) OnStop() {} + +func (s *selfRemovingSubscriber) len() int { + s.mu.Lock() + defer s.mu.Unlock() + return len(s.msgs) +} + +func (s *selfRemovingSubscriber) markerAt(idx int) byte { + s.mu.Lock() + defer s.mu.Unlock() + return payloadMarker(s.msgs[idx]) +} + func videoSeqHeader(marker byte) base.RtmpMsg { return base.RtmpMsg{ Header: base.RtmpHeader{MsgTypeId: base.RtmpTypeIdVideo}, @@ -168,8 +201,13 @@ func payloadMarker(msg base.RtmpMsg) byte { return msg.Payload[len(msg.Payload)-1] } +func newTestGroup(streamName string) *Group { + group, _ := GetGroupManagerInstance().GetOrCreateGroupByStreamName(streamName, streamName, nil, 1, 0) + return group +} + func TestAddConsumerReplaysCachedGopImmediately(t *testing.T) { - group := NewGroupByStreamName("test-replay", "test-replay", nil, 1, 0) + group := newTestGroup("test-replay") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-replay") group.OnMsg(videoSeqHeader(1)) @@ -194,7 +232,7 @@ func TestAddConsumerReplaysCachedGopImmediately(t *testing.T) { } func TestVideoSeqHeaderChangeClearsStaleGop(t *testing.T) { - group := NewGroupByStreamName("test-clear", "test-clear", nil, 1, 0) + group := newTestGroup("test-clear") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-clear") group.OnMsg(videoSeqHeader(1)) @@ -221,7 +259,7 @@ func TestVideoSeqHeaderChangeClearsStaleGop(t *testing.T) { } func TestNonAacAudioIsNotReplayedAsHeader(t *testing.T) { - group := NewGroupByStreamName("test-g711", "test-g711", nil, 1, 0) + group := newTestGroup("test-g711") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-g711") group.OnMsg(videoSeqHeader(1)) @@ -244,7 +282,7 @@ func TestNonAacAudioIsNotReplayedAsHeader(t *testing.T) { } func TestAddConsumerWithReplayDisabledDoesNotReplayCachedGop(t *testing.T) { - group := NewGroupByStreamName("test-no-replay", "test-no-replay", nil, 1, 0) + group := newTestGroup("test-no-replay") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-no-replay") group.OnMsg(videoSeqHeader(1)) @@ -276,7 +314,7 @@ func TestAddConsumerWithReplayDisabledDoesNotReplayCachedGop(t *testing.T) { } func TestAddConsumerReplayDoesNotInterleaveWithLiveKeyFrame(t *testing.T) { - group := NewGroupByStreamName("test-replay-order", "test-replay-order", nil, 1, 0) + group := newTestGroup("test-replay-order") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-replay-order") group.OnMsg(videoSeqHeader(1)) @@ -320,11 +358,54 @@ func TestAddConsumerReplayDoesNotInterleaveWithLiveKeyFrame(t *testing.T) { } } +func TestSubscriberRemovingItselfStopsReplayDelivery(t *testing.T) { + group := newTestGroup("test-self-remove-replay") + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-self-remove-replay") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(videoKeyNalu(2)) + group.OnMsg(videoInterNalu(3)) + + sub := &selfRemovingSubscriber{group: group, id: "consumer"} + group.AddConsumer(sub.id, sub) + + if sub.len() != 1 { + t.Fatalf("messages after self remove = %d, want 1", sub.len()) + } + if got := sub.markerAt(0); got != 1 { + t.Fatalf("first marker = %d, want 1", got) + } + + group.OnMsg(videoKeyNalu(4)) + if sub.len() != 1 { + t.Fatalf("messages after live frame = %d, want 1", sub.len()) + } +} + +func TestSubscriberRemovingItselfStopsHeaderAndLiveDelivery(t *testing.T) { + group := newTestGroup("test-self-remove-live") + defer GetGroupManagerInstance().RemoveGroupByStreamName("test-self-remove-live") + + group.OnMsg(videoSeqHeader(1)) + group.OnMsg(aacSeqHeader(2)) + + sub := &selfRemovingSubscriber{group: group, id: "consumer"} + group.AddConsumerWithReplay(sub.id, sub, false) + + group.OnMsg(videoKeyNalu(3)) + if sub.len() != 1 { + t.Fatalf("messages after self remove = %d, want 1", sub.len()) + } + if got := sub.markerAt(0); got != 1 { + t.Fatalf("first marker = %d, want 1", got) + } +} + func TestGroupManagerSupportsAppNameAndStreamName(t *testing.T) { manager := NewComplexGroupManager() group := &Group{key: NewStreamKey("live", "camera")} - manager.SetGroup(group.Key(), group) + manager.setGroup(group.Key(), group) ok, got := manager.GetGroup(NewStreamKey("live", "camera")) if !ok || got != group { @@ -339,8 +420,8 @@ func TestGroupManagerSupportsAppNameAndStreamName(t *testing.T) { func TestGroupManagerStreamNameFallbackRejectsAmbiguousAppName(t *testing.T) { manager := NewComplexGroupManager() - manager.SetGroup(NewStreamKey("app1", "camera"), &Group{key: NewStreamKey("app1", "camera")}) - manager.SetGroup(NewStreamKey("app2", "camera"), &Group{key: NewStreamKey("app2", "camera")}) + manager.setGroup(NewStreamKey("app1", "camera"), &Group{key: NewStreamKey("app1", "camera")}) + manager.setGroup(NewStreamKey("app2", "camera"), &Group{key: NewStreamKey("app2", "camera")}) ok, got := manager.GetGroup(StreamKeyFromStreamName("camera")) if ok || got != nil { @@ -348,14 +429,105 @@ func TestGroupManagerStreamNameFallbackRejectsAmbiguousAppName(t *testing.T) { } } +func TestGroupManagerGetOrCreateGroupReturnsExisting(t *testing.T) { + manager := NewComplexGroupManager() + key := NewStreamKey("live", "camera") + + group, created := manager.GetOrCreateGroup(key, "first", nil, 1, 0) + if !created || group == nil { + t.Fatal("expected group to be created") + } + + got, created := manager.GetOrCreateGroup(key, "second", nil, 1, 0) + if created || got != group { + t.Fatal("expected existing group to be returned") + } + if got.UniqueKey() != "first" { + t.Fatalf("unique key = %s, want first", got.UniqueKey()) + } +} + +func TestGroupManagerGetOrCreateWaitsForClosedGroupCleanup(t *testing.T) { + manager := NewComplexGroupManager() + key := StreamKeyFromStreamName("camera") + oldGroup := &Group{key: key} + oldGroup.closed.Store(true) + manager.setGroup(key, oldGroup) + + oldGroup.lifecycleMux.Lock() + done := make(chan struct { + group *Group + created bool + }) + go func() { + group, created := manager.GetOrCreateGroup(key, "new", nil, 1, 0) + done <- struct { + group *Group + created bool + }{group: group, created: created} + }() + + select { + case <-done: + t.Fatal("new group should wait for old group cleanup") + case <-time.After(50 * time.Millisecond): + } + + oldGroup.lifecycleMux.Unlock() + + select { + case result := <-done: + if !result.created || result.group == nil || result.group == oldGroup { + t.Fatalf("unexpected group result: group=%p created=%v", result.group, result.created) + } + case <-time.After(time.Second): + t.Fatal("new group was not created after old group cleanup") + } +} + +func TestGroupManagerGetOrCreateReturnsReplacementAfterWaitingClosedGroup(t *testing.T) { + manager := NewComplexGroupManager() + key := StreamKeyFromStreamName("camera") + oldGroup := &Group{key: key} + replacement := &Group{key: key} + oldGroup.closed.Store(true) + manager.setGroup(key, oldGroup) + + oldGroup.lifecycleMux.Lock() + done := make(chan struct { + group *Group + created bool + }) + go func() { + group, created := manager.GetOrCreateGroup(key, "new", nil, 1, 0) + done <- struct { + group *Group + created bool + }{group: group, created: created} + }() + + time.Sleep(50 * time.Millisecond) + manager.setGroup(key, replacement) + oldGroup.lifecycleMux.Unlock() + + select { + case result := <-done: + if result.created || result.group != replacement { + t.Fatalf("unexpected group result: group=%p replacement=%p created=%v", result.group, replacement, result.created) + } + case <-time.After(time.Second): + t.Fatal("replacement group was not returned after old group cleanup") + } +} + func TestGroupManagerRemoveGroupIfMatchDoesNotRemoveNewGroup(t *testing.T) { manager := NewComplexGroupManager() key := StreamKeyFromStreamName("camera") oldGroup := &Group{key: key} newGroup := &Group{key: key} - manager.SetGroup(key, oldGroup) - manager.SetGroup(key, newGroup) + manager.setGroup(key, oldGroup) + manager.setGroup(key, newGroup) manager.RemoveGroupIfMatch(key, oldGroup) ok, got := manager.GetGroup(key) @@ -370,12 +542,12 @@ func TestGroupManagerIterateRemoveDoesNotRemoveReplacement(t *testing.T) { oldGroup := &Group{key: key} newGroup := &Group{key: key} - manager.SetGroup(key, oldGroup) + manager.setGroup(key, oldGroup) manager.Iterate(func(iterKey StreamKey, group *Group) bool { if iterKey != key || group != oldGroup { t.Fatalf("unexpected iterate entry, key=%v group=%p", iterKey, group) } - manager.SetGroup(key, newGroup) + manager.setGroup(key, newGroup) return false }) @@ -415,7 +587,7 @@ func TestGopCacheNegativeFrameLimitMeansUnlimited(t *testing.T) { } func TestOnStopIsIdempotentAndClosesSubscribers(t *testing.T) { - group := NewGroupByStreamName("test-stop", "test-stop", nil, 1, 0) + group := newTestGroup("test-stop") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-stop") sub := &recordSubscriber{} @@ -435,7 +607,7 @@ func TestOnStopIsIdempotentAndClosesSubscribers(t *testing.T) { } func TestAddSubscriberAfterStopIsIgnored(t *testing.T) { - group := NewGroupByStreamName("test-add-after-stop", "test-add-after-stop", nil, 1, 0) + group := newTestGroup("test-add-after-stop") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-add-after-stop") group.OnStop() @@ -453,7 +625,7 @@ func TestAddSubscriberAfterStopIsIgnored(t *testing.T) { } func TestDuplicateSubscriberIDIsIgnored(t *testing.T) { - group := NewGroupByStreamName("test-duplicate", "test-duplicate", nil, 1, 0) + group := newTestGroup("test-duplicate") defer GetGroupManagerInstance().RemoveGroupByStreamName("test-duplicate") first := &recordSubscriber{} diff --git a/server/logs/lalserver.log b/server/logs/lalserver.log new file mode 100644 index 0000000..e0043cb --- /dev/null +++ b/server/logs/lalserver.log @@ -0,0 +1,210 @@ +2026/04/23 10:34:35.896771  INFO initial log succ. - config.go:249 +2026/04/23 10:34:35.896771  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:34:35.896771  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:34:35.896771  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:34:35.900773  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:34:35.900773  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:34:35.900773  INFO  start: 2026-04-23 10:34:35.866 - base.go:35 +2026/04/23 10:34:35.900773  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:34:35.900773  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build490025007\b363\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build490025007\b363\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:34:35.900773  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:34:35.900773  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:34:35.901773  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:34:35.901773  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:34:35.901773  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:34:35.901773  INFO [GROUP1] lifecycle new group. group=0xc000091888, appName=, streamName=test - group__.go:185 +2026/04/23 10:34:35.901773  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:34:35.901773 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:34:35.901773  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:34:35.902774  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:34:35.902774  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:34:35.902774 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:34:35.902774  INFO [GROUP2] lifecycle new group. group=0xc000091c08, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:34:35.902774  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:34:35.902774 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:34:35.902774  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:34:35.902774  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:34:35.902774  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:34:38.903343  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:34:38.904493  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 +2026/04/23 10:37:23.154712  INFO initial log succ. - config.go:249 +2026/04/23 10:37:23.155229  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:37:23.155229  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:37:23.155229  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:37:23.155229  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:37:23.155229  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:37:23.155229  INFO  start: 2026-04-23 10:37:23.133 - base.go:35 +2026/04/23 10:37:23.155229  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:37:23.155229  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build1390121107\b363\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build1390121107\b363\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:37:23.155229  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:37:23.155229  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:37:23.155229  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:37:23.155229  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:37:23.155229  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:37:23.155739  INFO [GROUP1] lifecycle new group. group=0xc000091888, appName=, streamName=test - group__.go:185 +2026/04/23 10:37:23.155739  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:37:23.155739 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:37:23.155739  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:37:23.155739  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:37:23.156259  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:37:23.156259  INFO [GROUP2] lifecycle new group. group=0xc000091c08, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:37:23.156259  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:37:23.156259 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:37:23.156259  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:37:23.156259  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:37:23.156259  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:37:23.156259 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:37:26.156615  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:37:26.156615  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 +2026/04/23 10:39:11.914284  INFO initial log succ. - config.go:249 +2026/04/23 10:39:11.914284  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:39:11.914284  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:39:11.914284  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:39:11.914284  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:39:11.914284  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:39:11.914284  INFO  start: 2026-04-23 10:39:11.897 - base.go:35 +2026/04/23 10:39:11.915156  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:39:11.915156  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build2037119015\b363\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build2037119015\b363\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:39:11.915156  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:39:11.915156  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:39:11.915156  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:39:11.915156  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:39:11.915156  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:39:11.915156  INFO [GROUP1] lifecycle new group. group=0xc0002e0708, appName=, streamName=test - group__.go:185 +2026/04/23 10:39:11.915156  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:39:11.915156 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:39:11.915156  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:39:11.915156  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:39:11.915912  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:39:11.915912  INFO [GROUP2] lifecycle new group. group=0xc0002e0a88, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:39:11.915912  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:39:11.915912 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:39:11.915912  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:39:11.915912  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:39:11.915912  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:39:11.915912 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:39:14.915985  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:39:14.916290  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 +2026/04/23 10:43:21.196284  INFO initial log succ. - config.go:249 +2026/04/23 10:43:21.196284  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:43:21.196284  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:43:21.196284  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:43:21.196284  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:43:21.196284  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:43:21.196284  INFO  start: 2026-04-23 10:43:21.177 - base.go:35 +2026/04/23 10:43:21.196284  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:43:21.196284  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build3820063322\b363\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build3820063322\b363\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:43:21.196284  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:43:21.196284  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:43:21.196284  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:43:21.196284  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:43:21.196284  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:43:21.196284  INFO [GROUP1] lifecycle new group. group=0xc000091888, appName=, streamName=test - group__.go:185 +2026/04/23 10:43:21.196284  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:43:21.196284 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:43:21.196284  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:43:21.196284  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:43:21.197284  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:43:21.197284  INFO [GROUP2] lifecycle new group. group=0xc000091c08, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:43:21.197284  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:43:21.197284 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:43:21.197284  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:43:21.197284  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:43:21.197284  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:43:21.197284 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:43:24.197564  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:43:24.197820  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 +2026/04/23 10:46:51.244739  INFO initial log succ. - config.go:249 +2026/04/23 10:46:51.244739  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:46:51.244739  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:46:51.244739  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:46:51.244739  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:46:51.244739  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:46:51.244739  INFO  start: 2026-04-23 10:46:51.225 - base.go:35 +2026/04/23 10:46:51.244739  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:46:51.244739  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build710474692\b363\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build710474692\b363\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:46:51.244739  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:46:51.244739  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:46:51.244739  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:46:51.244739  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:46:51.245259  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:46:51.245259  INFO [GROUP1] lifecycle new group. group=0xc000113888, appName=, streamName=test - group__.go:185 +2026/04/23 10:46:51.245259  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:46:51.245259 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:46:51.245259  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:46:51.245259  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:46:51.245779  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:46:51.245779  INFO [GROUP2] lifecycle new group. group=0xc000113c08, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:46:51.245779  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:46:51.245779 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:46:51.245779  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:46:51.245779  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:46:51.245779  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:46:51.246301 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:46:54.246259  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:46:54.246564  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 +2026/04/23 10:56:11.867506  INFO initial log succ. - config.go:249 +2026/04/23 10:56:11.867506  INFO  + __ ___ __ + / / / | / / + / / / /| | / / + / /___/ ___ |/ /___ +/_____/_/ |_/_____/ + - config.go:252 +2026/04/23 10:56:11.867506  WARN config version invalid. conf version of lalserver=v0.4.1, conf version of config file= - config.go:262 +2026/04/23 10:56:11.867506  WARN config some fields do not exist which have been set to the zero value. fields=[conf_version rtmp.addr rtmp.rtmps_enable rtmp.rtmps_addr rtmp.rtmps_cert_file rtmp.rtmps_key_file rtmp.gop_num rtmp.single_gop_max_frame_num rtmp.merge_write_size in_session.add_dummy_audio_enable in_session.add_dummy_audio_wait_audio_ms httpflv.enable httpflv.enable_https httpflv.url_pattern httpflv.gop_num httpflv.single_gop_max_frame_num hls.enable hls.enable_https hls.url_pattern hls.use_memory_as_disk_flag hls.out_path hls.fragment_duration_ms hls.fragment_num hls.delete_threshold hls.cleanup_mode hls.sub_session_timeout_ms hls.sub_session_hash_key httpts.enable httpts.enable_https httpts.url_pattern httpts.gop_num httpts.single_gop_max_frame_num rtsp.addr rtsp.rtsps_enable rtsp.rtsps_addr rtsp.rtsps_cert_file rtsp.rtsps_key_file rtsp.out_wait_key_frame_flag rtsp.ws_rtsp_enable rtsp.ws_rtsp_addr rtsp.auth_enable rtsp.auth_method rtsp.username rtsp.password record.enable_flv record.flv_out_path record.enable_mpegts record.mpegts_out_path relay_push.enable relay_push.addr_list static_relay_pull.enable static_relay_pull.addr http_api.addr server_id http_notify.enable http_notify.update_interval_sec http_notify.on_server_start http_notify.on_update http_notify.on_pub_start http_notify.on_pub_stop http_notify.on_sub_start http_notify.on_sub_stop http_notify.on_relay_pull_start http_notify.on_relay_pull_stop http_notify.on_rtmp_connect http_notify.on_hls_make_ts simple_auth.key simple_auth.dangerous_lal_secret simple_auth.pub_rtmp_enable simple_auth.sub_rtmp_enable simple_auth.sub_httpflv_enable simple_auth.sub_httpts_enable simple_auth.pub_rtsp_enable simple_auth.sub_rtsp_enable simple_auth.hls_m3u8_enable pprof.addr debug.log_group_interval_sec debug.log_group_max_group_num debug.log_group_max_sub_num_per_group] - config.go:278 +2026/04/23 10:56:11.867506  WARN config some log fields do not exist which have been set to default value. log.level=LevelDebug, log.filename=./logs/lalserver.log, log.is_to_stdout=true, log.is_rotate_daily=true, log.short_file_flag=true, log.timestamp_flag=true, log.timestamp_with_ms_flag=true, log.level_flag=true, log.assert_behavior=AssertError - config.go:283 +2026/04/23 10:56:11.867506  INFO load conf succ. raw content={"rtmp":{"enable":false},"rtsp":{"enable":false},"http_api":{"enable":false},"pprof":{"enable":false}} parsed=&{ConfVersion: RtmpConfig:{Enable:false Addr: RtmpsEnable:false RtmpsAddr: RtmpsCertFile: RtmpsKeyFile: GopNum:0 SingleGopMaxFrameNum:0 MergeWriteSize:0} InSessionConfig:{AddDummyAudioEnable:false AddDummyAudioWaitAudioMs:0} DefaultHttpConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:}} HttpflvConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} HlsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} UseMemoryAsDiskFlag:false MuxerConfig:{OutPath: FragmentDurationMs:0 FragmentNum:0 DeleteThreshold:0 CleanupMode:0} SubSessionTimeoutMs:0 SubSessionHashKey:} HttptsConfig:{CommonHttpServerConfig:{CommonHttpAddrConfig:{HttpListenAddr: HttpsListenAddr: HttpsCertFile: HttpsKeyFile:} Enable:false EnableHttps:false UrlPattern:} GopNum:0 SingleGopMaxFrameNum:0} RtspConfig:{Enable:false Addr: RtspsEnable:false RtspsAddr: RtspsCertFile: RtspsKeyFile: OutWaitKeyFrameFlag:false WsRtspEnable:false WsRtspAddr: ServerAuthConfig:{AuthEnable:false AuthMethod:0 UserName: PassWord:}} RecordConfig:{EnableFlv:false FlvOutPath: EnableMpegts:false MpegtsOutPath:} RelayPushConfig:{Enable:false AddrList:[]} StaticRelayPullConfig:{Enable:false Addr:} HttpApiConfig:{Enable:false Addr:} ServerId: HttpNotifyConfig:{Enable:false UpdateIntervalSec:0 OnServerStart: OnUpdate: OnPubStart: OnPubStop: OnSubStart: OnSubStop: OnRelayPullStart: OnRelayPullStop: OnRtmpConnect: OnHlsMakeTs:} SimpleAuthConfig:{Key: DangerousLalSecret: PubRtmpEnable:false SubRtmpEnable:false SubHttpflvEnable:false SubHttptsEnable:false PubRtspEnable:false SubRtspEnable:false HlsM3u8Enable:false} PprofConfig:{Enable:false Addr:} LogConfig:{Level:1 Filename:./logs/lalserver.log IsToStdout:true IsRotateDaily:true IsRotateHourly:false ShortFileFlag:true TimestampFlag:true TimestampWithMsFlag:true LevelFlag:true AssertBehavior:1 HookBackendOutFn:} DebugConfig:{LogGroupIntervalSec:0 LogGroupMaxGroupNum:0 LogGroupMaxSubNumPerGroup:0}} - config.go:346 +2026/04/23 10:56:11.867506  INFO  start: 2026-04-23 10:56:11.849 - base.go:35 +2026/04/23 10:56:11.867506  INFO  wd: F:\开源\go\lalmi\lalmax\server - base.go:36 +2026/04/23 10:56:11.867506  INFO  args: C:\Users\ADMINI~1\AppData\Local\Temp\go-build3560235915\b364\server.test.exe -test.testlogfile=C:\Users\ADMINI~1\AppData\Local\Temp\go-build3560235915\b364\testlog.txt -test.paniconexit0 -test.timeout=10m0s - base.go:37 +2026/04/23 10:56:11.867506  INFO  bininfo: GitTag=unknown. GitCommitLog=unknown. GitStatus=unknown. BuildTime=unknown. GoVersion=unknown. runtime=windows/amd64. - base.go:38 +2026/04/23 10:56:11.867506  INFO  version: lal v0.37.4 (github.com/q191201771/lal) - base.go:39 +2026/04/23 10:56:11.867506  INFO  github: https://github.com/q191201771/lal - base.go:40 +2026/04/23 10:56:11.867506  INFO  doc: https://pengrl.com/lal - base.go:41 +2026/04/23 10:56:11.867506  INFO lalmax http listen. addr=:52349 - server.go:103 +2026/04/23 10:56:11.867506  INFO [GROUP1] lifecycle new group. group=0xc000091888, appName=, streamName=test - group__.go:185 +2026/04/23 10:56:11.867506  INFO [CUSTOMIZEPUB1] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:56:11.867506 DEBUG [GROUP1] [CUSTOMIZEPUB1] add customize pub session into group. - group__in.go:33 +2026/04/23 10:56:11.867506  INFO create group, uniqueKey:, streamKey:test - group.go:108 +2026/04/23 10:56:11.867506  INFO SetGroup, streamKey:test - group_manager.go:105 +2026/04/23 10:56:11.868976  INFO AddSubscriber, streamKey:test, subscriberId:consumer1, protocol:LALMAX - group.go:237 +2026/04/23 10:56:11.868976  INFO [GROUP2] lifecycle new group. group=0xc000091c08, appName=, streamName=notify_test - group__.go:185 +2026/04/23 10:56:11.868976  INFO [CUSTOMIZEPUB2] NewCustomizePubSessionContext. - customize_pubsession.go:42 +2026/04/23 10:56:11.868976 DEBUG [GROUP2] [CUSTOMIZEPUB2] add customize pub session into group. - group__in.go:33 +2026/04/23 10:56:11.868976  INFO create group, uniqueKey:, streamKey:notify_test - group.go:108 +2026/04/23 10:56:11.868976  INFO SetGroup, streamKey:notify_test - group_manager.go:105 +2026/04/23 10:56:11.868976  INFO AddSubscriber, streamKey:notify_test, subscriberId:consumer_notify, protocol:LALMAX - group.go:237 +2026/04/23 10:56:11.868976 ERROR http notify post error. err=Post "http://127.0.0.1:55559/on_update": dial tcp 127.0.0.1:55559: connectex: No connection could be made because the target machine actively refused it., url=http://127.0.0.1:55559/on_update, info={EventCommonInfo:{ServerId:} Groups:[]} - http_notify.go:216 +2026/04/23 10:56:14.869938  INFO http api start rtp pub. req info={StreamName:rtp_pub_test Port:0 TimeoutMs:0 IsTcpFlag:0 DebugDumpPacket:} - router.go:245 +2026/04/23 10:56:14.870245  INFO http api stop rtp pub. stream_name=rtp_pub_test, session_id= - router.go:271 diff --git a/server/router.go b/server/router.go index 954182e..f941696 100644 --- a/server/router.go +++ b/server/router.go @@ -7,8 +7,6 @@ import ( maxlogic "github.com/q191201771/lalmax/logic" - "github.com/q191201771/lalmax/gb28181" - "github.com/gin-gonic/gin" "github.com/q191201771/lal/pkg/base" "github.com/q191201771/lal/pkg/logic" @@ -41,19 +39,6 @@ func (s *LalMaxServer) InitRouter(router *gin.Engine) { // hls-fmp4/llhls router.GET("/live/hls/:streamid/:type", s.HandleHls) - // gb - gbLogic := gb28181.NewGbLogic(s.gbsbr) - gb := router.Group("/api/gb") - gb.GET("/device_infos", gbLogic.GetDeviceInfos) - gb.POST("/start_play", gbLogic.StartPlay) - gb.POST("/stop_play", gbLogic.StopPlay) - gb.POST("/update_all_notify", gbLogic.UpdateAllNotify) - gb.POST("/update_notify", gbLogic.UpdateNotify) - gb.POST("/ptz_direction", gbLogic.PtzDirection) - gb.POST("/ptz_zoom", gbLogic.PtzZoom) - gb.POST("/ptz_fi", gbLogic.PtzFi) - gb.POST("/ptz_preset", gbLogic.PtzPreset) - gb.POST("/ptz_stop", gbLogic.PtzStop) auth := Authentication(s.conf.HttpConfig.CtrlAuthWhitelist.Secrets, s.conf.HttpConfig.CtrlAuthWhitelist.IPs) // stat @@ -68,6 +53,7 @@ func (s *LalMaxServer) InitRouter(router *gin.Engine) { ctrl.POST("/stop_relay_pull", s.ctrlStopRelayPullHandler) ctrl.POST("/kick_session", s.ctrlKickSessionHandler) ctrl.POST("/start_rtp_pub", s.ctrlStartRtpPubHandler) + ctrl.POST("/stop_rtp_pub", s.ctrlStopRtpPubHandler) } func (s *LalMaxServer) HandleWHIP(c *gin.Context) { @@ -258,11 +244,46 @@ func (s *LalMaxServer) ctrlStartRtpPubHandler(c *gin.Context) { Log.Infof("http api start rtp pub. req info=%+v", info) - lal := s.lalsvr.(*logic.ServerManager) - resp := lal.CtrlStartRtpPub(info) + resp := s.rtpPubMgr.Start(info) c.JSON(http.StatusOK, resp) } +func (s *LalMaxServer) ctrlStopRtpPubHandler(c *gin.Context) { + var v base.ApiCtrlStopRelayPullResp + streamName := c.Query("stream_name") + sessionID := c.Query("session_id") + + if streamName == "" && sessionID == "" { + var info base.ApiCtrlKickSessionReq + if _, err := unmarshalRequestJSONBody(c.Request, &info); err == nil { + streamName = info.StreamName + sessionID = info.SessionId + } + } + + if streamName == "" && sessionID == "" { + v.ErrorCode = base.ErrorCodeParamMissing + v.Desp = base.DespParamMissing + c.JSON(http.StatusOK, v) + return + } + + Log.Infof("http api stop rtp pub. stream_name=%s, session_id=%s", streamName, sessionID) + + session, err := s.rtpPubMgr.Stop(streamName, sessionID) + if err != nil { + v.ErrorCode = base.ErrorCodeSessionNotFound + v.Desp = err.Error() + c.JSON(http.StatusOK, v) + return + } + + v.ErrorCode = base.ErrorCodeSucc + v.Desp = base.DespSucc + v.Data.SessionId = session.ID + c.JSON(http.StatusOK, v) +} + func unmarshalRequestJSONBody(r *http.Request, info interface{}, keyFieldList ...string) (nazajson.Json, error) { body, err := io.ReadAll(r.Body) if err != nil { diff --git a/server/router_test.go b/server/router_test.go index 9c19d22..4df705e 100644 --- a/server/router_test.go +++ b/server/router_test.go @@ -1,6 +1,7 @@ package server import ( + "bytes" "encoding/json" "fmt" "net/http" @@ -69,9 +70,8 @@ func TestAllGroup(t *testing.T) { }) t.Run("has consumer", func(t *testing.T) { - ss := maxlogic.NewGroupByStreamName("test", "test", max.hlssvr, 1, 0) + ss, _ := maxlogic.GetGroupManagerInstance().GetOrCreateGroupByStreamName("test", "test", max.hlssvr, 1, 0) ss.AddConsumer("consumer1", nil) - maxlogic.GetGroupManagerInstance().SetGroupByStreamName("test", ss) r := httptest.NewRecorder() req := httptest.NewRequest("GET", "/api/stat/all_group", nil) @@ -105,9 +105,8 @@ func TestNotifyUpdate(t *testing.T) { if err != nil { t.Fatal(err) } - ss := maxlogic.NewGroupByStreamName(streamName, streamName, max.hlssvr, 1, 0) + ss, _ := maxlogic.GetGroupManagerInstance().GetOrCreateGroupByStreamName(streamName, streamName, max.hlssvr, 1, 0) ss.AddConsumer(consumerID, nil) - maxlogic.GetGroupManagerInstance().SetGroupByStreamName(streamName, ss) http.HandleFunc("/on_update", func(w http.ResponseWriter, r *http.Request) { var out base.ApiStatAllGroupResp @@ -127,6 +126,47 @@ func TestNotifyUpdate(t *testing.T) { time.Sleep(time.Second * 3) } +func TestRtpPubStartStop(t *testing.T) { + body := bytes.NewBufferString(`{"stream_name":"rtp_pub_test","port":0,"timeout_ms":0}`) + r := httptest.NewRecorder() + req := httptest.NewRequest("POST", "/api/ctrl/start_rtp_pub", body) + max.router.ServeHTTP(r, req) + resp := r.Result() + if resp.StatusCode != http.StatusOK { + t.Fatal(resp.Status) + } + + var startResp base.ApiCtrlStartRtpPubResp + if err := json.NewDecoder(resp.Body).Decode(&startResp); err != nil { + t.Fatal(err) + } + if startResp.ErrorCode != base.ErrorCodeSucc { + t.Fatalf("start_rtp_pub failed, code=%d desp=%s", startResp.ErrorCode, startResp.Desp) + } + if startResp.Data.StreamName != "rtp_pub_test" || startResp.Data.SessionId == "" || startResp.Data.Port == 0 { + t.Fatalf("unexpected start_rtp_pub data: %+v", startResp.Data) + } + + r = httptest.NewRecorder() + req = httptest.NewRequest("POST", "/api/ctrl/stop_rtp_pub?stream_name=rtp_pub_test", nil) + max.router.ServeHTTP(r, req) + resp = r.Result() + if resp.StatusCode != http.StatusOK { + t.Fatal(resp.Status) + } + + var stopResp base.ApiCtrlStopRelayPullResp + if err := json.NewDecoder(resp.Body).Decode(&stopResp); err != nil { + t.Fatal(err) + } + if stopResp.ErrorCode != base.ErrorCodeSucc { + t.Fatalf("stop_rtp_pub failed, code=%d desp=%s", stopResp.ErrorCode, stopResp.Desp) + } + if stopResp.Data.SessionId != startResp.Data.SessionId { + t.Fatalf("stop_rtp_pub session id = %s, want %s", stopResp.Data.SessionId, startResp.Data.SessionId) + } +} + func TestAuthentication(t *testing.T) { t.Run("无须鉴权", func(t *testing.T) { if !authentication("12", "192.168.0.2", nil, nil) { diff --git a/server/server.go b/server/server.go index 7a28353..466c8ec 100644 --- a/server/server.go +++ b/server/server.go @@ -9,9 +9,9 @@ import ( "github.com/q191201771/lalmax/rtc" - maxlogic "github.com/q191201771/lalmax/logic" + "github.com/q191201771/lalmax/gb28181/rtppub" - "github.com/q191201771/lalmax/gb28181" + maxlogic "github.com/q191201771/lalmax/logic" httpfmp4 "github.com/q191201771/lalmax/fmp4/http-fmp4" @@ -33,7 +33,7 @@ type LalMaxServer struct { routerTls *gin.Engine httpfmp4svr *httpfmp4.HttpFmp4Server hlssvr *hls.HlsServer - gbsbr *gb28181.GB28181Server + rtpPubMgr *rtppub.Manager } func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { @@ -47,8 +47,9 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { }) maxsvr := &LalMaxServer{ - lalsvr: lalsvr, - conf: conf, + lalsvr: lalsvr, + conf: conf, + rtpPubMgr: rtppub.NewManager(lalsvr, conf.GB28181Config.MediaConfig), } if conf.SrtConfig.Enable { @@ -75,10 +76,6 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { maxsvr.hlssvr = hls.NewHlsServer(conf.Fmp4Config.Hls) } - if conf.GB28181Config.Enable { - maxsvr.gbsbr = gb28181.NewGB28181Server(conf.GB28181Config, lalsvr) - } - maxsvr.router = gin.Default() maxsvr.InitRouter(maxsvr.router) if conf.HttpConfig.EnableHttps { @@ -91,8 +88,8 @@ func NewLalMaxServer(conf *config.Config) (*LalMaxServer, error) { func (s *LalMaxServer) Run() (err error) { s.lalsvr.WithOnHookSession(func(uniqueKey string, streamName string) logic.ICustomizeHookSessionContext { - // lal 有新的输入流时，创建 lalmax 扩展流组用于分发扩展协议。 - return maxlogic.NewGroupByStreamName(uniqueKey, streamName, s.hlssvr, s.conf.LogicConfig.GopCacheNum, s.conf.LogicConfig.SingleGopMaxFrameNum) + group, _ := maxlogic.GetGroupManagerInstance().GetOrCreateGroupByStreamName(uniqueKey, streamName, s.hlssvr, s.conf.LogicConfig.GopCacheNum, s.conf.LogicConfig.SingleGopMaxFrameNum) + return group }) ctx, cancel := context.WithCancel(context.Background()) @@ -119,9 +116,5 @@ func (s *LalMaxServer) Run() (err error) { }() } - if s.gbsbr != nil { - go s.gbsbr.Start() - } - return s.lalsvr.RunLoop() }