diff --git a/internal/connrtmp/conn.go b/internal/connrtmp/conn.go index bcee7b6e..502eb3b8 100644 --- a/internal/connrtmp/conn.go +++ b/internal/connrtmp/conn.go @@ -73,11 +73,12 @@ type Conn struct { pathMan PathMan parent Parent - // read mode - ringBuffer *ringbuffer.RingBuffer + path readpublisher.Path + ringBuffer *ringbuffer.RingBuffer // read // in - terminate chan struct{} + terminate chan struct{} + parentTerminate chan struct{} } // New allocates a Conn. @@ -106,7 +107,8 @@ func New( conn: rtmp.NewServerConn(nconn), pathMan: pathMan, parent: parent, - terminate: make(chan struct{}), + terminate: make(chan struct{}, 1), + parentTerminate: make(chan struct{}), } c.log(logger.Info, "opened") @@ -117,15 +119,18 @@ func New( return c } -// Close closes a Conn. -func (c *Conn) Close() { +// ParentClose closes a Conn. +func (c *Conn) ParentClose() { c.log(logger.Info, "closed") - close(c.terminate) + close(c.parentTerminate) } -// RequestClose closes a Conn. -func (c *Conn) RequestClose() { - c.parent.OnConnClose(c) +// Close closes a Conn. +func (c *Conn) Close() { + select { + case c.terminate <- struct{}{}: + default: + } } // IsReadPublisher implements readpublisher.ReadPublisher. @@ -154,443 +159,352 @@ func (c *Conn) run() { defer onConnectCmd.Close() } - c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) - c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) - err := c.conn.ServerHandshake() - if err != nil { - c.log(logger.Info, "ERR: %s", err) + c.ringBuffer = ringbuffer.New(uint64(c.readBufferCount)) + + connErr := make(chan error) + go func() { + connErr <- func() error { + c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) + c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) + err := c.conn.ServerHandshake() + if err != nil { + return err + } + + if c.conn.IsPublishing() { + return c.runPublish() + } + return c.runRead() + }() + }() + + select { + case err := <-connErr: + if err != io.EOF { + c.log(logger.Info, "ERR: %s", err) + } + c.conn.NetConn().Close() - c.parent.OnConnClose(c) - <-c.terminate - return + case <-c.terminate: + c.ringBuffer.Close() + c.conn.NetConn().Close() + <-connErr } - if c.conn.IsPublishing() { - c.runPublish() - } else { - c.runRead() + if c.path != nil { + res := make(chan struct{}) + c.path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet + <-res } + + c.parent.OnConnClose(c) + <-c.parentTerminate } -func (c *Conn) runRead() { - var path readpublisher.Path +func (c *Conn) runRead() error { + pathName, query := pathNameAndQuery(c.conn.URL()) + + sres := make(chan readpublisher.SetupPlayRes) + c.pathMan.OnReadPublisherSetupPlay(readpublisher.SetupPlayReq{ + Author: c, + PathName: pathName, + IP: c.ip(), + ValidateCredentials: func(authMethods []headers.AuthMethod, pathUser string, pathPass string) error { + return c.validateCredentials(pathUser, pathPass, query) + }, + Res: sres}) + res := <-sres + + if res.Err != nil { + if _, ok := res.Err.(readpublisher.ErrAuthCritical); ok { + // wait some seconds to stop brute force attacks + <-time.After(pauseAfterAuthError) + } + return res.Err + } + + c.path = res.Path + var videoTrack *gortsplib.Track var h264Decoder *rtph264.Decoder var audioTrack *gortsplib.Track var audioClockRate int var aacDecoder *rtpaac.Decoder - err := func() error { - pathName, query := pathNameAndQuery(c.conn.URL()) - - sres := make(chan readpublisher.SetupPlayRes) - c.pathMan.OnReadPublisherSetupPlay(readpublisher.SetupPlayReq{ - Author: c, - PathName: pathName, - IP: c.ip(), - ValidateCredentials: func(authMethods []headers.AuthMethod, pathUser string, pathPass string) error { - return c.validateCredentials(pathUser, pathPass, query) - }, - Res: sres}) - res := <-sres - - if res.Err != nil { - if _, ok := res.Err.(readpublisher.ErrAuthCritical); ok { - // wait some seconds to stop brute force attacks - select { - case <-time.After(pauseAfterAuthError): - case <-c.terminate: - } + for i, t := range res.Tracks { + if t.IsH264() { + if videoTrack != nil { + return fmt.Errorf("can't read track %d with RTMP: too many tracks", i+1) } - return res.Err - } - path = res.Path + videoTrack = t + h264Decoder = rtph264.NewDecoder() - for i, t := range res.Tracks { - if t.IsH264() { - if videoTrack != nil { - return fmt.Errorf("can't read track %d with RTMP: too many tracks", i+1) - } - - videoTrack = t - h264Decoder = rtph264.NewDecoder() - - } else if t.IsAAC() { - if audioTrack != nil { - return fmt.Errorf("can't read track %d with RTMP: too many tracks", i+1) - } - - audioTrack = t - audioClockRate, _ = audioTrack.ClockRate() - aacDecoder = rtpaac.NewDecoder(audioClockRate) + } else if t.IsAAC() { + if audioTrack != nil { + return fmt.Errorf("can't read track %d with RTMP: too many tracks", i+1) } + + audioTrack = t + audioClockRate, _ = audioTrack.ClockRate() + aacDecoder = rtpaac.NewDecoder(audioClockRate) } - - if videoTrack == nil && audioTrack == nil { - return fmt.Errorf("unable to find a video or audio track") - } - - c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) - c.conn.WriteMetadata(videoTrack, audioTrack) - - return nil - }() - if err != nil { - c.conn.NetConn().Close() - c.log(logger.Info, "ERR: %v", err) - - if path != nil { - res := make(chan struct{}) - path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet - <-res - } - - c.parent.OnConnClose(c) - <-c.terminate - return } - c.ringBuffer = ringbuffer.New(uint64(c.readBufferCount)) + if videoTrack == nil && audioTrack == nil { + return fmt.Errorf("unable to find a video or audio track") + } + + c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) + c.conn.WriteMetadata(videoTrack, audioTrack) pres := make(chan readpublisher.PlayRes) - path.OnReadPublisherPlay(readpublisher.PlayReq{c, pres}) //nolint:govet + c.path.OnReadPublisherPlay(readpublisher.PlayReq{c, pres}) //nolint:govet <-pres - c.log(logger.Info, "is reading from path '%s'", path.Name()) + c.log(logger.Info, "is reading from path '%s'", c.path.Name()) // disable read deadline c.conn.NetConn().SetReadDeadline(time.Time{}) - writerDone := make(chan error) - go func() { - writerDone <- func() error { - var videoBuf [][]byte - videoDTSEst := h264.NewDTSEstimator() + var videoBuf [][]byte + videoDTSEst := h264.NewDTSEstimator() - for { - data, ok := c.ringBuffer.Pull() - if !ok { - return fmt.Errorf("terminated") - } - pair := data.(trackIDPayloadPair) - - if videoTrack != nil && pair.trackID == videoTrack.ID { - nalus, pts, err := h264Decoder.Decode(pair.buf) - if err != nil { - if err != rtph264.ErrMorePacketsNeeded { - c.log(logger.Warn, "unable to decode video track: %v", err) - } - continue - } - - for _, nalu := range nalus { - // remove SPS, PPS and AUD, not needed by RTSP - typ := h264.NALUType(nalu[0] & 0x1F) - switch typ { - case h264.NALUTypeSPS, h264.NALUTypePPS, h264.NALUTypeAccessUnitDelimiter: - continue - } - - videoBuf = append(videoBuf, nalu) - } - - // RTP marker means that all the NALUs with the same PTS have been received. - // send them together. - marker := (pair.buf[1] >> 7 & 0x1) > 0 - if marker { - data, err := h264.EncodeAVCC(videoBuf) - if err != nil { - return err - } - - dts := videoDTSEst.Feed(pts + ptsOffset) - c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) - err = c.conn.WritePacket(av.Packet{ - Type: av.H264, - Data: data, - Time: dts, - CTime: pts + ptsOffset - dts, - }) - if err != nil { - return err - } - - videoBuf = nil - } - - } else if audioTrack != nil && pair.trackID == audioTrack.ID { - aus, pts, err := aacDecoder.Decode(pair.buf) - if err != nil { - if err != rtpaac.ErrMorePacketsNeeded { - c.log(logger.Warn, "unable to decode audio track: %v", err) - } - continue - } - - for i, au := range aus { - auPTS := pts + ptsOffset + time.Duration(i)*1000*time.Second/time.Duration(audioClockRate) - - c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) - err := c.conn.WritePacket(av.Packet{ - Type: av.AAC, - Data: au, - Time: auPTS, - }) - if err != nil { - return err - } - } - } - } - }() - }() - - select { - case err := <-writerDone: - c.conn.NetConn().Close() - - if err != io.EOF { - c.log(logger.Info, "ERR: %s", err) + for { + data, ok := c.ringBuffer.Pull() + if !ok { + return fmt.Errorf("terminated") } + pair := data.(trackIDPayloadPair) - res := make(chan struct{}) - path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet - <-res - - c.parent.OnConnClose(c) - <-c.terminate - - case <-c.terminate: - res := make(chan struct{}) - path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet - <-res - - c.ringBuffer.Close() - c.conn.NetConn().Close() - <-writerDone - } -} - -func (c *Conn) runPublish() { - var videoTrack *gortsplib.Track - var audioTrack *gortsplib.Track - var err error - var tracks gortsplib.Tracks - var h264Encoder *rtph264.Encoder - var aacEncoder *rtpaac.Encoder - var path readpublisher.Path - - setupDone := make(chan struct{}) - go func() { - defer close(setupDone) - err = func() error { - c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) - videoTrack, audioTrack, err = c.conn.ReadMetadata() + if videoTrack != nil && pair.trackID == videoTrack.ID { + nalus, pts, err := h264Decoder.Decode(pair.buf) if err != nil { - return err - } - - if videoTrack != nil { - h264Encoder = rtph264.NewEncoder(96, nil, nil, nil) - tracks = append(tracks, videoTrack) - } - - if audioTrack != nil { - clockRate, _ := audioTrack.ClockRate() - aacEncoder = rtpaac.NewEncoder(96, clockRate, nil, nil, nil) - tracks = append(tracks, audioTrack) - } - - for i, t := range tracks { - t.ID = i - } - - pathName, query := pathNameAndQuery(c.conn.URL()) - - resc := make(chan readpublisher.AnnounceRes) - c.pathMan.OnReadPublisherAnnounce(readpublisher.AnnounceReq{ - Author: c, - PathName: pathName, - Tracks: tracks, - IP: c.ip(), - ValidateCredentials: func(authMethods []headers.AuthMethod, pathUser string, pathPass string) error { - return c.validateCredentials(pathUser, pathPass, query) - }, - Res: resc, - }) - res := <-resc - - if res.Err != nil { - if _, ok := res.Err.(readpublisher.ErrAuthCritical); ok { - // wait some seconds to stop brute force attacks - select { - case <-time.After(pauseAfterAuthError): - case <-c.terminate: - } + if err != rtph264.ErrMorePacketsNeeded { + c.log(logger.Warn, "unable to decode video track: %v", err) } - return res.Err + continue } - path = res.Path - return nil - }() - }() - - select { - case <-setupDone: - case <-c.terminate: - c.conn.NetConn().Close() - <-setupDone - } - - if err != nil { - c.conn.NetConn().Close() - c.log(logger.Info, "ERR: %s", err) - - c.parent.OnConnClose(c) - <-c.terminate - return - } - - // disable write deadline - c.conn.NetConn().SetWriteDeadline(time.Time{}) - - readerDone := make(chan error) - go func() { - readerDone <- func() error { - resc := make(chan readpublisher.RecordRes) - path.OnReadPublisherRecord(readpublisher.RecordReq{Author: c, Res: resc}) - res := <-resc - - if res.Err != nil { - return res.Err - } - - c.log(logger.Info, "is publishing to path '%s', %d %s", - path.Name(), - len(tracks), - func() string { - if len(tracks) == 1 { - return "track" - } - return "tracks" - }()) - - var onPublishCmd *externalcmd.Cmd - if path.Conf().RunOnPublish != "" { - _, port, _ := net.SplitHostPort(c.rtspAddress) - onPublishCmd = externalcmd.New(path.Conf().RunOnPublish, - path.Conf().RunOnPublishRestart, externalcmd.Environment{ - Path: path.Name(), - Port: port, - }) - } - - defer func(path readpublisher.Path) { - if path.Conf().RunOnPublish != "" { - onPublishCmd.Close() + for _, nalu := range nalus { + // remove SPS, PPS and AUD, not needed by RTSP + typ := h264.NALUType(nalu[0] & 0x1F) + switch typ { + case h264.NALUTypeSPS, h264.NALUTypePPS, h264.NALUTypeAccessUnitDelimiter: + continue } - }(path) - rtcpSenders := rtcpsenderset.New(tracks, res.SP.OnFrame) - defer rtcpSenders.Close() - - onFrame := func(trackID int, payload []byte) { - rtcpSenders.OnFrame(trackID, gortsplib.StreamTypeRTP, payload) - res.SP.OnFrame(trackID, gortsplib.StreamTypeRTP, payload) + videoBuf = append(videoBuf, nalu) } - for { - c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) - pkt, err := c.conn.ReadPacket() + // RTP marker means that all the NALUs with the same PTS have been received. + // send them together. + marker := (pair.buf[1] >> 7 & 0x1) > 0 + if marker { + data, err := h264.EncodeAVCC(videoBuf) if err != nil { return err } - switch pkt.Type { - case av.H264: - if videoTrack == nil { - return fmt.Errorf("ERR: received an H264 frame, but track is not set up") - } + dts := videoDTSEst.Feed(pts + ptsOffset) + c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) + err = c.conn.WritePacket(av.Packet{ + Type: av.H264, + Data: data, + Time: dts, + CTime: pts + ptsOffset - dts, + }) + if err != nil { + return err + } - nalus, err := h264.DecodeAVCC(pkt.Data) - if err != nil { - return err - } + videoBuf = nil + } - var outNALUs [][]byte + } else if audioTrack != nil && pair.trackID == audioTrack.ID { + aus, pts, err := aacDecoder.Decode(pair.buf) + if err != nil { + if err != rtpaac.ErrMorePacketsNeeded { + c.log(logger.Warn, "unable to decode audio track: %v", err) + } + continue + } - for _, nalu := range nalus { - // remove SPS, PPS and AUD, not needed by RTSP - typ := h264.NALUType(nalu[0] & 0x1F) - switch typ { - case h264.NALUTypeSPS, h264.NALUTypePPS, h264.NALUTypeAccessUnitDelimiter: - continue - } + for i, au := range aus { + auPTS := pts + ptsOffset + time.Duration(i)*1000*time.Second/time.Duration(audioClockRate) - outNALUs = append(outNALUs, nalu) - } - - if len(outNALUs) == 0 { - continue - } - - frames, err := h264Encoder.Encode(outNALUs, pkt.Time+pkt.CTime) - if err != nil { - return fmt.Errorf("ERR while encoding H264: %v", err) - } - - for _, frame := range frames { - onFrame(videoTrack.ID, frame) - } - - case av.AAC: - if audioTrack == nil { - return fmt.Errorf("ERR: received an AAC frame, but track is not set up") - } - - frames, err := aacEncoder.Encode([][]byte{pkt.Data}, pkt.Time+pkt.CTime) - if err != nil { - return fmt.Errorf("ERR while encoding AAC: %v", err) - } - - for _, frame := range frames { - onFrame(audioTrack.ID, frame) - } - - default: - return fmt.Errorf("ERR: unexpected packet: %v", pkt.Type) + c.conn.NetConn().SetWriteDeadline(time.Now().Add(c.writeTimeout)) + err := c.conn.WritePacket(av.Packet{ + Type: av.AAC, + Data: au, + Time: auPTS, + }) + if err != nil { + return err } } - }() - }() + } + } +} - select { - case err := <-readerDone: - c.conn.NetConn().Close() +func (c *Conn) runPublish() error { + c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) + videoTrack, audioTrack, err := c.conn.ReadMetadata() + if err != nil { + return err + } - if err != io.EOF { - c.log(logger.Info, "ERR: %s", err) + var tracks gortsplib.Tracks + var h264Encoder *rtph264.Encoder + var aacEncoder *rtpaac.Encoder + + if videoTrack != nil { + h264Encoder = rtph264.NewEncoder(96, nil, nil, nil) + tracks = append(tracks, videoTrack) + } + + if audioTrack != nil { + clockRate, _ := audioTrack.ClockRate() + aacEncoder = rtpaac.NewEncoder(96, clockRate, nil, nil, nil) + tracks = append(tracks, audioTrack) + } + + for i, t := range tracks { + t.ID = i + } + + pathName, query := pathNameAndQuery(c.conn.URL()) + + resc := make(chan readpublisher.AnnounceRes) + c.pathMan.OnReadPublisherAnnounce(readpublisher.AnnounceReq{ + Author: c, + PathName: pathName, + Tracks: tracks, + IP: c.ip(), + ValidateCredentials: func(authMethods []headers.AuthMethod, pathUser string, pathPass string) error { + return c.validateCredentials(pathUser, pathPass, query) + }, + Res: resc, + }) + res := <-resc + + if res.Err != nil { + if _, ok := res.Err.(readpublisher.ErrAuthCritical); ok { + // wait some seconds to stop brute force attacks + <-time.After(pauseAfterAuthError) + } + return res.Err + } + + c.path = res.Path + + // disable write deadline + c.conn.NetConn().SetWriteDeadline(time.Time{}) + + rresc := make(chan readpublisher.RecordRes) + c.path.OnReadPublisherRecord(readpublisher.RecordReq{Author: c, Res: rresc}) + rres := <-rresc + + if rres.Err != nil { + return rres.Err + } + + c.log(logger.Info, "is publishing to path '%s', %d %s", + c.path.Name(), + len(tracks), + func() string { + if len(tracks) == 1 { + return "track" + } + return "tracks" + }()) + + var onPublishCmd *externalcmd.Cmd + if c.path.Conf().RunOnPublish != "" { + _, port, _ := net.SplitHostPort(c.rtspAddress) + onPublishCmd = externalcmd.New(c.path.Conf().RunOnPublish, + c.path.Conf().RunOnPublishRestart, externalcmd.Environment{ + Path: c.path.Name(), + Port: port, + }) + } + + defer func(path readpublisher.Path) { + if path.Conf().RunOnPublish != "" { + onPublishCmd.Close() + } + }(c.path) + + rtcpSenders := rtcpsenderset.New(tracks, rres.SP.OnFrame) + defer rtcpSenders.Close() + + onFrame := func(trackID int, payload []byte) { + rtcpSenders.OnFrame(trackID, gortsplib.StreamTypeRTP, payload) + rres.SP.OnFrame(trackID, gortsplib.StreamTypeRTP, payload) + } + + for { + c.conn.NetConn().SetReadDeadline(time.Now().Add(c.readTimeout)) + pkt, err := c.conn.ReadPacket() + if err != nil { + return err } - res := make(chan struct{}) - path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet - <-res - path = nil + switch pkt.Type { + case av.H264: + if videoTrack == nil { + return fmt.Errorf("ERR: received an H264 frame, but track is not set up") + } - c.parent.OnConnClose(c) - <-c.terminate + nalus, err := h264.DecodeAVCC(pkt.Data) + if err != nil { + return err + } - case <-c.terminate: - c.conn.NetConn().Close() - <-readerDone + var outNALUs [][]byte - res := make(chan struct{}) - path.OnReadPublisherRemove(readpublisher.RemoveReq{c, res}) //nolint:govet - <-res - path = nil + for _, nalu := range nalus { + // remove SPS, PPS and AUD, not needed by RTSP + typ := h264.NALUType(nalu[0] & 0x1F) + switch typ { + case h264.NALUTypeSPS, h264.NALUTypePPS, h264.NALUTypeAccessUnitDelimiter: + continue + } + + outNALUs = append(outNALUs, nalu) + } + + if len(outNALUs) == 0 { + continue + } + + frames, err := h264Encoder.Encode(outNALUs, pkt.Time+pkt.CTime) + if err != nil { + return fmt.Errorf("ERR while encoding H264: %v", err) + } + + for _, frame := range frames { + onFrame(videoTrack.ID, frame) + } + + case av.AAC: + if audioTrack == nil { + return fmt.Errorf("ERR: received an AAC frame, but track is not set up") + } + + frames, err := aacEncoder.Encode([][]byte{pkt.Data}, pkt.Time+pkt.CTime) + if err != nil { + return fmt.Errorf("ERR while encoding AAC: %v", err) + } + + for _, frame := range frames { + onFrame(audioTrack.ID, frame) + } + + default: + return fmt.Errorf("ERR: unexpected packet: %v", pkt.Type) + } } } diff --git a/internal/connrtsp/conn.go b/internal/connrtsp/conn.go index 3f5ab53c..64b5a892 100644 --- a/internal/connrtsp/conn.go +++ b/internal/connrtsp/conn.go @@ -95,8 +95,8 @@ func New( return c } -// Close closes a Conn. -func (c *Conn) Close(err error) { +// ParentClose closes a Conn. +func (c *Conn) ParentClose(err error) { if err != io.EOF && !isTeardownErr(err) && !isTerminatedErr(err) { c.log(logger.Info, "ERR: %v", err) } diff --git a/internal/converterhls/converter.go b/internal/converterhls/converter.go index f9af53ad..7a816fed 100644 --- a/internal/converterhls/converter.go +++ b/internal/converterhls/converter.go @@ -182,14 +182,14 @@ func New( return c } -// Close closes a Converter. -func (c *Converter) Close() { +// ParentClose closes a Converter. +func (c *Converter) ParentClose() { c.log(logger.Info, "closed") close(c.terminate) } -// RequestClose closes a Converter. -func (c *Converter) RequestClose() { +// Close closes a Converter. +func (c *Converter) Close() { c.parent.OnConverterClose(c) } diff --git a/internal/path/path.go b/internal/path/path.go index 744b0866..46e8f67b 100644 --- a/internal/path/path.go +++ b/internal/path/path.go @@ -309,7 +309,7 @@ outer: case readPublisherStateRecord: atomic.AddInt64(pa.stats.CountPublishers, -1) } - c.RequestClose() + c.Close() } } pa.readPublishersWg.Wait() @@ -470,7 +470,7 @@ func (pa *Path) removeReadPublisher(c readpublisher.ReadPublisher) { for oc, state := range pa.readPublishers { if state != readPublisherStatePreRemove { pa.removeReadPublisher(oc) - oc.RequestClose() + oc.Close() } } } @@ -510,7 +510,7 @@ func (pa *Path) onSourceSetNotReady() { for c, state := range pa.readPublishers { if c != pa.source && state != readPublisherStatePreRemove { pa.removeReadPublisher(c) - c.RequestClose() + c.Close() } } } @@ -661,10 +661,10 @@ func (pa *Path) onReadPublisherAnnounce(req readpublisher.AnnounceReq) { return } - pa.Log(logger.Info, "disconnecting existing publisher") + pa.Log(logger.Info, "closing existing publisher") curPublisher := pa.source.(readpublisher.ReadPublisher) pa.removeReadPublisher(curPublisher) - curPublisher.RequestClose() + curPublisher.Close() // prevent path closure if pa.closeTimerStarted { diff --git a/internal/readpublisher/readpublisher.go b/internal/readpublisher/readpublisher.go index 81774b4e..5d51d996 100644 --- a/internal/readpublisher/readpublisher.go +++ b/internal/readpublisher/readpublisher.go @@ -58,7 +58,6 @@ type ReadPublisher interface { IsReadPublisher() IsSource() Close() - RequestClose() OnFrame(int, gortsplib.StreamType, []byte) } diff --git a/internal/serverhls/server.go b/internal/serverhls/server.go index f0dbf756..75ab3511 100644 --- a/internal/serverhls/server.go +++ b/internal/serverhls/server.go @@ -203,7 +203,7 @@ func (s *Server) ServeHTTP(w http.ResponseWriter, r *http.Request) { func (s *Server) doConverterClose(c *converterhls.Converter) { delete(s.converters, c.PathName()) - c.Close() + c.ParentClose() } // OnConverterClose is called by converterhls.Converter. diff --git a/internal/serverrtmp/server.go b/internal/serverrtmp/server.go index 37211277..227fdc24 100644 --- a/internal/serverrtmp/server.go +++ b/internal/serverrtmp/server.go @@ -183,6 +183,7 @@ outer: func (s *Server) doConnClose(c *connrtmp.Conn) { delete(s.conns, c) + c.ParentClose() c.Close() } diff --git a/internal/serverrtsp/server.go b/internal/serverrtsp/server.go index 3d190849..ad823960 100644 --- a/internal/serverrtsp/server.go +++ b/internal/serverrtsp/server.go @@ -221,7 +221,7 @@ func (s *Server) OnConnClose(ctx *gortsplib.ServerHandlerOnConnCloseCtx) { delete(s.conns, ctx.Conn) s.mutex.Unlock() - c.Close(ctx.Error) + c.ParentClose(ctx.Error) } // OnRequest implements gortsplib.ServerHandlerOnRequest. @@ -270,7 +270,7 @@ func (s *Server) OnSessionClose(ctx *gortsplib.ServerHandlerOnSessionCloseCtx) { delete(s.sessions, ctx.Session) s.mutex.Unlock() - se.Close() + se.ParentClose() } // OnDescribe implements gortsplib.ServerHandlerOnDescribe. diff --git a/internal/sessionrtsp/session.go b/internal/sessionrtsp/session.go index a3e2d7f2..1144325b 100644 --- a/internal/sessionrtsp/session.go +++ b/internal/sessionrtsp/session.go @@ -75,8 +75,8 @@ func New( return s } -// Close closes a Session. -func (s *Session) Close() { +// ParentClose closes a Session. +func (s *Session) ParentClose() { switch s.ss.State() { case gortsplib.ServerSessionStatePlay: if s.onReadCmd != nil { @@ -99,8 +99,8 @@ func (s *Session) Close() { s.log(logger.Info, "closed") } -// RequestClose closes a Session. -func (s *Session) RequestClose() { +// Close closes a Session. +func (s *Session) Close() { s.ss.Close() }