[opt] customize: 当DelCustomizePubSession后，调用被删除对象的FeedAvPacket方法将返回错误

pkg/base/avpacket_stream.go

@@ -46,7 +46,7 @@ type IAvPacketStream interface {
 	// 注意，调用 FeedAvPacket 传入AAC音频数据前，需要先调用 FeedAudioSpecificConfig。
 	// FeedAudioSpecificConfig 在最开始总共调用一次，后面就可以一直调用 FeedAvPacket
 	//
-	FeedAudioSpecificConfig(asc []byte)
+	FeedAudioSpecificConfig(asc []byte) error
 
 	// FeedAvPacket
 	//
@@ -66,5 +66,5 @@ type IAvPacketStream interface {
 	// Avcc也即[<4字节长度 + nal>...]，Annexb也即[<4字节start code 00 00 00 01 + nal>...]。
 	// 注意，sps和pps也通过 FeedAvPacket 传入。sps和pps可以单独调用 FeedAvPacket，也可以sps+pps+I帧组合在一起调用一次 FeedAvPacket
 	//
-	FeedAvPacket(packet AvPacket)
+	FeedAvPacket(packet AvPacket) error
 }

pkg/base/error.go

@@ -76,7 +76,8 @@ var ErrSdp = errors.New("lal.sdp: fxxk")
 // ----- pkg/logic -----------------------------------------------------------------------------------------------------
 
 var (
-	ErrDupInStream = errors.New("lal.logic: in stream already exist at group")
+	ErrDupInStream      = errors.New("lal.logic: in stream already exist at group")
+	ErrDisposedInStream = errors.New("lal.logic: in stream already disposed")
 
 	ErrSimpleAuthParamNotFound = errors.New("lal.logic: simple auth failed since url param lal_secret not found")
 	ErrSimpleAuthFailed        = errors.New("lal.logic: simple auth failed since url param lal_secret invalid")

pkg/logic/customize_pubsession.go

@@ -11,6 +11,7 @@ package logic
 import (
 	"github.com/q191201771/lal/pkg/base"
 	"github.com/q191201771/lal/pkg/remux"
+	"github.com/q191201771/naza/pkg/nazaatomic"
 )
 
 type CustomizePubSessionContext struct {
@@ -19,6 +20,8 @@ type CustomizePubSessionContext struct {
 	streamName string
 	remuxer    *remux.AvPacket2RtmpRemuxer
 	onRtmpMsg  func(msg base.RtmpMsg)
+
+	disposeFlag nazaatomic.Bool
 }
 
 func NewCustomizePubSessionContext(streamName string) *CustomizePubSessionContext {
@@ -42,16 +45,28 @@ func (ctx *CustomizePubSessionContext) StreamName() string {
 	return ctx.streamName
 }
 
+func (ctx *CustomizePubSessionContext) Dispose() {
+	ctx.disposeFlag.Store(true)
+}
+
 // ---------------------------------------------------------------------------------------------------------------------
 
 func (ctx *CustomizePubSessionContext) WithOption(modOption func(option *base.AvPacketStreamOption)) {
 	ctx.remuxer.WithOption(modOption)
 }
 
-func (ctx *CustomizePubSessionContext) FeedAudioSpecificConfig(asc []byte) {
+func (ctx *CustomizePubSessionContext) FeedAudioSpecificConfig(asc []byte) error {
+	if ctx.disposeFlag.Load() {
+		return base.ErrDisposedInStream
+	}
 	ctx.remuxer.InitWithAvConfig(asc, nil, nil, nil)
+	return nil
 }
 
-func (ctx *CustomizePubSessionContext) FeedAvPacket(packet base.AvPacket) {
+func (ctx *CustomizePubSessionContext) FeedAvPacket(packet base.AvPacket) error {
+	if ctx.disposeFlag.Load() {
+		return base.ErrDisposedInStream
+	}
 	ctx.remuxer.FeedAvPacket(packet)
+	return nil
 }

pkg/logic/logic.go

@@ -27,7 +27,9 @@ type ILalServer interface {
 	//
 	AddCustomizePubSession(streamName string) (ICustomizePubSessionContext, error)
 
-	// DelCustomizePubSession 将 ICustomizePubSessionContext 从 ILalServer 中删除
+	// DelCustomizePubSession 将 ICustomizePubSessionContext 对象从 ILalServer 中删除
+	//
+	// 注意，业务方调用该函数后，就不要再使用该 ICustomizePubSessionContext 对象的方法了，比如继续 FeedAvPacket 是无效的
 	//
 	DelCustomizePubSession(ICustomizePubSessionContext)