From 0ed0d75a2be15f99e54160f2d464f40a0fb94133 Mon Sep 17 00:00:00 2001
From: Jae-Sung Lee <jaesung.lee9507@gmail.com>
Date: Thu, 17 Nov 2022 07:36:04 +0000
Subject: [PATCH] Add example flv file of CustomPubSession

---
 .../customize_lalserver.go                    | 39 +++++++++++++++++--
 pkg/logic/customize_pubsession.go             |  9 +++++
 pkg/logic/logic.go                            |  5 ++-
 3 files changed, 49 insertions(+), 4 deletions(-)

diff --git a/app/demo/customize_lalserver/customize_lalserver.go b/app/demo/customize_lalserver/customize_lalserver.go
index 0d44456..5ab1f5c 100644
--- a/app/demo/customize_lalserver/customize_lalserver.go
+++ b/app/demo/customize_lalserver/customize_lalserver.go
@@ -11,13 +11,16 @@ package main
 import (
 	"flag"
 	"fmt"
-	"github.com/q191201771/lal/pkg/aac"
-	"github.com/q191201771/lal/pkg/avc"
-	"github.com/q191201771/naza/pkg/nazalog"
 	"io/ioutil"
 	"os"
 	"time"
 
+	"github.com/q191201771/lal/pkg/aac"
+	"github.com/q191201771/lal/pkg/avc"
+	"github.com/q191201771/lal/pkg/httpflv"
+	"github.com/q191201771/lal/pkg/remux"
+	"github.com/q191201771/naza/pkg/nazalog"
+
 	"github.com/q191201771/lal/pkg/base"
 
 	"github.com/q191201771/lal/pkg/logic"
@@ -38,6 +41,7 @@ func main() {
 
 	// 比常规lalserver多加了这一行
 	go showHowToCustomizePub(lals)
+	go showHowToFlvCustomizePub(lals)
 
 	err := lals.RunLoop()
 	nazalog.Infof("server manager done. err=%+v", err)
@@ -57,6 +61,35 @@ func parseFlag() string {
 	return *cf
 }
 
+func showHowToFlvCustomizePub(lals logic.ILalServer) {
+	const (
+		flvfilename            = "/tmp/test.flv"
+		customizePubStreamName = "f110"
+	)
+
+	time.Sleep(200 * time.Millisecond)
+
+	tags, err := httpflv.ReadAllTagsFromFlvFile(flvfilename)
+	nazalog.Assert(nil, err)
+
+	session, err := lals.AddCustomizePubSession(customizePubStreamName)
+	nazalog.Assert(nil, err)
+
+	startRealTime := time.Now()
+	startTs := int64(0)
+	for _, tag := range tags {
+		msg := remux.FlvTag2RtmpMsg(tag)
+		diffTs := int64(msg.Header.TimestampAbs) - startTs
+		diffReal := time.Since(startRealTime).Milliseconds()
+		if diffReal < diffTs {
+			time.Sleep(time.Duration(diffTs-diffReal) * time.Millisecond)
+		}
+		session.FeedRtmpMsg(msg)
+	}
+
+	lals.DelCustomizePubSession(session)
+}
+
 func showHowToCustomizePub(lals logic.ILalServer) {
 	const (
 		h264filename = "/tmp/test.h264"
diff --git a/pkg/logic/customize_pubsession.go b/pkg/logic/customize_pubsession.go
index 2cbeea4..cc35016 100644
--- a/pkg/logic/customize_pubsession.go
+++ b/pkg/logic/customize_pubsession.go
@@ -36,6 +36,7 @@ func NewCustomizePubSessionContext(streamName string) *CustomizePubSessionContex
 }
 
 func (ctx *CustomizePubSessionContext) WithOnRtmpMsg(onRtmpMsg func(msg base.RtmpMsg)) *CustomizePubSessionContext {
+	ctx.onRtmpMsg = onRtmpMsg
 	ctx.remuxer.WithOnRtmpMsg(onRtmpMsg)
 	return ctx
 }
@@ -76,3 +77,11 @@ func (ctx *CustomizePubSessionContext) FeedAvPacket(packet base.AvPacket) error
 	ctx.remuxer.FeedAvPacket(packet)
 	return nil
 }
+
+func (ctx *CustomizePubSessionContext) FeedRtmpMsg(msg base.RtmpMsg) error {
+	if ctx.disposeFlag.Load() {
+		return base.ErrDisposedInStream
+	}
+	ctx.onRtmpMsg(msg)
+	return nil
+}
diff --git a/pkg/logic/logic.go b/pkg/logic/logic.go
index bee6fb1..9aa3231 100644
--- a/pkg/logic/logic.go
+++ b/pkg/logic/logic.go
@@ -9,8 +9,9 @@
 package logic
 
 import (
-	"github.com/q191201771/lal/pkg/base"
 	"path/filepath"
+
+	"github.com/q191201771/lal/pkg/base"
 )
 
 // ---------------------------------------------------------------------------------------------------------------------
@@ -60,6 +61,8 @@ type ICustomizePubSessionContext interface {
 	//
 	base.IAvPacketStream
 
+	FeedRtmpMsg(msg base.RtmpMsg) error
+
 	UniqueKey() string
 	StreamName() string
 }