fix: delete device

api.go

@@ -747,7 +747,9 @@ func (s *Server) RemoveDevice(ctx context.Context, req *pb.RequestWithId) (res *
 		err = pkg.ErrNoDB
 		return
 	}
-	tx := s.DB.Delete(&Device{}, req.Id)
+	tx := s.DB.Delete(&Device{
+		ID: uint(req.Id),
+	})
 	err = tx.Error
 	s.Devices.Call(func() error {
 		if device, ok := s.Devices.Get(uint(req.Id)); ok {

example/8080/gb28181_server.yaml

@@ -0,0 +1,8 @@
+gb28181:
+  autoinvite: true
+  sip:
+    listenaddr:
+          - udp::5060
+  onsub:
+    pull:
+      .* : $0

go.mod

@@ -119,6 +119,7 @@ require (
 	github.com/onsi/ginkgo/v2 v2.9.5 // indirect
 	github.com/phsym/console-slog v0.3.1
 	github.com/prometheus/client_golang v1.20.4
+	github.com/quangngotan95/go-m3u8 v0.1.0
 	go.uber.org/mock v0.4.0 // indirect
 	golang.org/x/crypto v0.26.0 // indirect
 	golang.org/x/exp v0.0.0-20240716175740-e3f259677ff7 // indirect

go.sum

@@ -251,6 +251,8 @@ github.com/prometheus/common v0.55.0 h1:KEi6DK7lXW/m7Ig5i47x0vRzuBsHuvJdi5ee6Y3G
 github.com/prometheus/common v0.55.0/go.mod h1:2SECS4xJG1kd8XF9IcM1gMX6510RAEL65zxzNImwdc8=
 github.com/prometheus/procfs v0.15.1 h1:YagwOFzUgYfKKHX6Dr+sHT7km/hxC76UB0learggepc=
 github.com/prometheus/procfs v0.15.1/go.mod h1:fB45yRUv8NstnjriLhBQLuOUt+WW4BsoGhij/e3PBqk=
+github.com/quangngotan95/go-m3u8 v0.1.0 h1:8oseBjJn5IKHQKdRZwSNskkua3NLrRtlvXXtoVgBzMk=
+github.com/quangngotan95/go-m3u8 v0.1.0/go.mod h1:smzfWHlYpBATVNu1GapKLYiCtEo5JxridIgvvudZ+Wc=
 github.com/quic-go/quic-go v0.43.1 h1:fLiMNfQVe9q2JvSsiXo4fXOEguXHGGl9+6gLp4RPeZQ=
 github.com/quic-go/quic-go v0.43.1/go.mod h1:132kz4kL3F9vxhW3CtQJLDVwcFe5wdWeJXXijhsO57M=
 github.com/rogpeppe/go-internal v1.12.0 h1:exVL4IDcn6na9z1rAb56Vxr+CgyK3nn3O+epU5NdKM8=

pkg/track.go

@@ -85,7 +85,7 @@ func (t *Track) AddBytesIn(n int) {
 }
 
 func (t *AVTrack) Ready(err error) {
-	if !t.IsReady() {
+	if t.ready.IsPending() {
 		if err != nil {
 			t.Error("ready", "err", err)
 		} else {
@@ -101,7 +101,7 @@ func (t *AVTrack) Ready(err error) {
 }
 
 func (t *Track) Ready(err error) {
-	if !t.IsReady() {
+	if t.ready.IsPending() {
 		if err != nil {
 			t.Error("ready", "err", err)
 		} else {

plugin/hls/index.go

@@ -0,0 +1,12 @@
+package plugin_hls
+
+import (
+	"m7s.live/v5"
+	"m7s.live/v5/plugin/hls/pkg"
+)
+
+var _ = m7s.InstallPlugin[HLSPlugin](hls.NewPuller)
+
+type HLSPlugin struct {
+	m7s.Plugin
+}

plugin/hls/pkg/m3u8.go

@@ -0,0 +1,54 @@
+package hls
+
+import (
+	"compress/gzip"
+	"io"
+	"net/http"
+	"net/url"
+	"sync"
+
+	"github.com/quangngotan95/go-m3u8/m3u8"
+)
+
+type M3u8Info struct {
+	Req       *http.Request
+	M3U8Count int    //一共拉取的m3u8文件数量
+	TSCount   int    //一共拉取的ts文件数量
+	LastM3u8  string //最后一个m3u8文件内容
+}
+
+type TSDownloader struct {
+	client *http.Client
+	url    *url.URL
+	req    *http.Request
+	res    *http.Response
+	wg     sync.WaitGroup
+	err    error
+	dur    float64
+}
+
+func (p *TSDownloader) Start() {
+	p.wg.Add(1)
+	go func() {
+		defer p.wg.Done()
+		if tsRes, err := p.client.Do(p.req); err == nil {
+			p.res = tsRes
+		} else {
+			p.err = err
+		}
+	}()
+}
+
+func readM3U8(res *http.Response) (playlist *m3u8.Playlist, err error) {
+	var reader io.Reader = res.Body
+	if res.Header.Get("Content-Encoding") == "gzip" {
+		reader, err = gzip.NewReader(reader)
+	}
+	if err == nil {
+		playlist, err = m3u8.Read(reader)
+	}
+	if err != nil {
+		// HLSPlugin.Error("readM3U8", zap.Error(err))
+	}
+	return
+}

plugin/hls/pkg/pull.go

@@ -0,0 +1,34 @@
+package hls
+
+import (
+	"context"
+	"net/http"
+	"sync"
+
+	"m7s.live/v5"
+	"m7s.live/v5/pkg/config"
+	"m7s.live/v5/pkg/task"
+)
+
+type Puller struct {
+	m7s.HTTPFilePuller
+	Video       M3u8Info
+	Audio       M3u8Info
+	TsHead      http.Header     `json:"-" yaml:"-"` //用于提供cookie等特殊身份的http头
+	SaveContext context.Context `json:"-" yaml:"-"` //用来保存ts文件到服务器
+	memoryTs    sync.Map
+}
+
+func NewPuller(_ config.Pull) m7s.IPuller {
+	p := &Puller{}
+	p.SetDescription(task.OwnerTypeKey, "HLSPuller")
+	return p
+}
+
+func (p *Puller) GetTs(key string) (any, bool) {
+	return p.memoryTs.Load(key)
+}
+
+func (p *Puller) Run() (err error) {
+	return
+}

publisher.go

@@ -200,8 +200,8 @@ func (p *Publisher) Start() (err error) {
 			}
 		}
 	}
-	p.audioReady = util.NewPromiseWithTimeout(p, time.Second*5)
-	p.videoReady = util.NewPromiseWithTimeout(p, time.Second*5)
+	p.audioReady = util.NewPromise(p)
+	p.videoReady = util.NewPromise(p)
 	if p.Dump {
 		f := filepath.Join("./dump", p.StreamPath)
 		os.MkdirAll(filepath.Dir(f), 0666)