fix: empty record file

example/8081/pull_rtsp.yaml

@@ -3,6 +3,12 @@ global:
   tcp: :50052
   http: :8081
   disableall: true
+flv:
+  enable: true
+preview:
+  enable: true
+console:
+  enable: true
 rtsp:
   enable: true
   tcp: :8554

plugin/gb28181/client.go

@@ -0,0 +1,108 @@
+package plugin_gb28181
+
+import (
+	"context"
+	"fmt"
+	"net"
+	"os"
+	"strconv"
+	"strings"
+	"time"
+
+	"github.com/emiago/sipgo"
+	"github.com/emiago/sipgo/sip"
+	"github.com/icholy/digest"
+	"github.com/rs/zerolog"
+	"m7s.live/pro/pkg/task"
+)
+
+type Client struct {
+	*sipgo.Client
+	task.Job
+	conf *GB28181Plugin
+}
+
+type KeepAlive struct {
+	task.Task
+	client *Client
+}
+
+func (c *Client) Start() (err error) {
+	netWork, parent, _ := strings.Cut(c.conf.Parent, ":")
+	host, portStr, _ := net.SplitHostPort(parent)
+	if portStr != "" {
+		portStr = "5060"
+	}
+	port, _ := strconv.Atoi(portStr)
+	c.Client, err = sipgo.NewClient(c.conf.ua, sipgo.WithClientLogger(zerolog.New(os.Stdout)), sipgo.WithClientHostname(host), sipgo.WithClientPort(port))
+	if err != nil {
+		return
+	}
+	recipient := sip.Uri{}
+	sip.ParseUri(fmt.Sprintf("sip:%s@%s", c.conf.Serial, c.conf.Realm), &recipient)
+	req := sip.NewRequest("REGISTER", recipient)
+	req.AppendHeader(
+		sip.NewHeader("Contact", fmt.Sprintf("<sip:%s@%s>", c.conf.Serial, c.conf.Realm)),
+	)
+	req.SetTransport(strings.ToUpper(netWork))
+	var res *sip.Response
+	res, err = c.Do(c.conf, req)
+	if err != nil {
+		c.conf.Error("register", "error", err.Error())
+	} else {
+		c.conf.Info("register", "response", res.String())
+		if res.StatusCode == 401 {
+			// Get WwW-Authenticate
+			wwwAuth := res.GetHeader("WWW-Authenticate")
+			var chal *digest.Challenge
+			chal, err = digest.ParseChallenge(wwwAuth.Value())
+			if err != nil {
+				c.conf.Error("register", "error", err.Error())
+			}
+
+			// Reply with digest
+			cred, _ := digest.Digest(chal, digest.Options{
+				Method:   req.Method.String(),
+				URI:      recipient.Host,
+				Username: c.conf.Username,
+				Password: c.conf.Password,
+			})
+
+			newReq := req.Clone()
+			newReq.RemoveHeader("Via") // Must be regenerated by tranport layer
+			newReq.AppendHeader(sip.NewHeader("Authorization", cred.String()))
+
+			ctx := context.Background()
+			var tx sip.ClientTransaction
+			tx, err = c.TransactionRequest(ctx, newReq, sipgo.ClientRequestAddVia)
+			if err != nil {
+				c.conf.Error("register", "error", err.Error())
+			}
+			defer tx.Terminate()
+
+			res, err = getResponse(tx)
+			if err != nil {
+				c.conf.Error("register", "error", err.Error())
+				return
+			}
+		}
+		var ka KeepAlive
+		ka.client = c
+		ka.SetRetry(-1, time.Second*10)
+		c.AddTask(&ka)
+	}
+	return
+}
+
+func getResponse(tx sip.ClientTransaction) (*sip.Response, error) {
+	select {
+	case <-tx.Done():
+		return nil, fmt.Errorf("transaction died")
+	case res := <-tx.Responses():
+		return res, nil
+	}
+}
+
+func (c *Client) Run() (err error) {
+	return
+}

plugin/gb28181/index.go

@@ -53,7 +53,6 @@ type GB28181Plugin struct {
 	Parent     string `desc:"父级设备"`
 	ua         *sipgo.UserAgent
 	server     *sipgo.Server
-	client     *sipgo.Client
 	devices    util.Collection[string, *Device]
 	dialogs    util.Collection[uint32, *Dialog]
 	tcpPorts   chan uint16
@@ -105,16 +104,10 @@ func (gb *GB28181Plugin) OnInit() (err error) {
 		}
 	}
 	if gb.Parent != "" {
-		host, portStr, _ := net.SplitHostPort(gb.Parent)
-		if portStr != "" {
-			portStr = "5060"
-		}
-		port, _ := strconv.Atoi(portStr)
-		gb.client, _ = sipgo.NewClient(gb.ua, sipgo.WithClientLogger(logger), sipgo.WithClientHostname(host), sipgo.WithClientPort(port))
-		gb.client.Do(gb, sip.NewRequest("REGISTER", sip.Uri{
-			Host: host,
-			Port: port,
-		}))
+		var client Client
+		client.conf = gb
+		client.SetRetry(-1, time.Second*5)
+		gb.AddTask(&client)
 	}
 	return
 }

publisher.go

@@ -10,6 +10,7 @@ import (
 	"sync"
 	"time"
 
+	"m7s.live/pro/pkg"
 	"m7s.live/pro/pkg/task"
 
 	. "m7s.live/pro/pkg"
@@ -660,7 +661,7 @@ func (p *Publisher) takeOver(old *Publisher) {
 }
 
 func (p *Publisher) WaitTrack() (err error) {
-	var v, a error
+	var v, a error = pkg.ErrNoTrack, pkg.ErrNoTrack
 	if p.PubVideo {
 		v = p.videoReady.Await()
 	}