mqtt加入认证，优化视频播放与浏览器

internal/common/pause_sub.go

@@ -1,45 +1,50 @@
 package common
 
-import "sync"
+import (
+	"context"
+	"sync"
+)
 
 type PauseSub struct {
-	ctrl *CtrlWait
-	// 回调函数
+	ctrl  *CtrlWait
 	items []chan int8
 	m     sync.RWMutex
 }
 
-// Add 添加一个暂停项
-func (p *PauseSub) Add(item chan int8) {
+func (p *PauseSub) GetNew() chan int8 {
 	p.m.Lock()
 	defer p.m.Unlock()
-	p.items = append(p.items, item)
+	sub := make(chan int8)
+	p.items = append(p.items, sub)
+	return sub
 }
 
-// Remove 移除一个暂停项
-func (p *PauseSub) Remove(item chan int8) {
+func (p *PauseSub) Close(sub chan int8) {
 	p.m.Lock()
 	defer p.m.Unlock()
+	close(sub)
 	for i, v := range p.items {
-		if v == item {
+		if v == sub {
 			p.items = append(p.items[:i], p.items[i+1:]...)
 		}
 	}
 }
 
 // Run 开始监听
-func (p *PauseSub) Run() {
+func (p *PauseSub) Run(ctx context.Context) {
 	p.ctrl.Open()
 	defer p.ctrl.Close()
 
 	for {
 		select {
-		case <-p.ctrl.C:
+		case <-ctx.Done():
+			return
+		case c := <-p.ctrl.C:
 			go func() {
 				p.m.RLock()
 				defer p.m.RUnlock()
 				for _, item := range p.items {
-					item <- 1
+					item <- c
 				}
 			}()
 		}
@@ -48,6 +53,7 @@ func (p *PauseSub) Run() {
 
 func NewPauseSub(c *CtrlWait) *PauseSub {
 	return &PauseSub{
-		ctrl: c,
+		ctrl:  c,
+		items: make([]chan int8, 0),
 	}
 }