Merge branch 'v4' of https://github.com/Monibuca/plugin-webrtc into v4

2 years ago · 806e713c56
8 changed files with 239 additions and 230 deletions
--- a/batcher.go
+++ b/batcher.go
@ -2,14 +2,9 @@ package webrtc

 import (
 	"encoding/json"
-	"fmt"
-	"time"

-	"github.com/pion/rtcp"
 	. "github.com/pion/webrtc/v3"
 	"go.uber.org/zap"
-	. "m7s.live/engine/v4"
-	. "m7s.live/engine/v4/track"
 )

 type Signal struct {
@ -19,17 +14,13 @@ type Signal struct {
 	Answer     string   `json:"answer"`
 	StreamPath string   `json:"streamPath"`
 }
-type BatchUplink struct {
-	Publisher
-	StreamPath string
-}
+
 type WebRTCBatcher struct {
-	WebRTCIO
 	PageSize      int
 	PageNum       int
 	subscribers   []*WebRTCBatchSubscriber
 	signalChannel *DataChannel
-	BatchUplink
+	WebRTCPublisher
 }

 func (suber *WebRTCBatcher) Start() (err error) {
@ -50,63 +41,14 @@ func (suber *WebRTCBatcher) Start() (err error) {
 		WebRTCPlugin.Info("Connection State has changed:" + pcs.String())
 		switch pcs {
 		case PeerConnectionStateConnected:
-			suber.OnTrack(func(track *TrackRemote, receiver *RTPReceiver) {
-				if suber.Publisher.Stream == nil {
-					WebRTCPlugin.Publish(suber.StreamPath, &suber.BatchUplink)
-				}
-				if suber.Publisher.Stream == nil {
-					return
-				}
-				puber := &suber.Publisher
-				if codec := track.Codec(); track.Kind() == RTPCodecTypeAudio {
-					if puber.AudioTrack == nil {
-						switch codec.PayloadType {
-						case 8:
-							puber.AudioTrack = NewG711(puber.Stream, true)
-						case 0:
-							puber.AudioTrack = NewG711(puber.Stream, false)
-						default:
-							puber.AudioTrack = nil
-							return
-						}
-					}
-					for {
-						b := make([]byte, 1460)
-						if i, _, err := track.Read(b); err == nil {
-							puber.AudioTrack.WriteRTP(b[:i])
-						} else {
-							return
-						}
-					}
-				} else {
-					go func() {
-						ticker := time.NewTicker(webrtcConfig.PLI)
-						for {
-							select {
-							case <-ticker.C:
-								if rtcpErr := suber.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}}); rtcpErr != nil {
-									fmt.Println(rtcpErr)
-								}
-							case <-puber.Done():
-								return
-							}
-						}
-					}()
-					puber.VideoTrack = NewH264(puber.Stream, byte(codec.PayloadType))
-					for {
-						b := make([]byte, 1460)
-						if i, _, err := track.Read(b); err == nil {
-							puber.VideoTrack.WriteRTP(b[:i])
-						} else {
-							return
-						}
-					}
-				}
-			})
+
 		case PeerConnectionStateDisconnected, PeerConnectionStateFailed:
 			for _, sub := range suber.subscribers {
 				go sub.Stop()
 			}
+			if suber.Publisher.Stream != nil {
+				suber.Publisher.Stop()
+			}
 			suber.PeerConnection.Close()
 		}
 	})
@ -121,8 +63,6 @@ func (suber *WebRTCBatcher) Signal(msg DataChannelMessage) {
 		WebRTCPlugin.Error("Signal", zap.Error(err))
 	} else {
 		switch s.Type {
-		case "streamPath":
-			suber.StreamPath = s.StreamPath
 		case "subscribe":
 			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: s.Offer}); err != nil {
 				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
@ -163,12 +103,13 @@ func (suber *WebRTCBatcher) Signal(msg DataChannelMessage) {
 				WebRTCPlugin.Error("Signal GetAnswer", zap.Error(err))
 				return
 			}
+
 		// if offer, err = suber.CreateOffer(nil); err == nil {
 		// 	b, _ := json.Marshal(offer)
 		// 	err = suber.signalChannel.SendText(string(b))
 		// 	suber.SetLocalDescription(offer)
 		// }
-		case "publish":
+		case "publish", "unpublish":
 			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: s.Offer}); err != nil {
 				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
 				return
@ -182,11 +123,18 @@ func (suber *WebRTCBatcher) Signal(msg DataChannelMessage) {
 				WebRTCPlugin.Error("Signal GetAnswer", zap.Error(err))
 				return
 			}
+			switch s.Type {
+			case "publish":
+				WebRTCPlugin.Publish(s.StreamPath, suber)
+			case "unpublish":
+				suber.Stop()
+			}
 		case "answer":
 			if err = suber.SetRemoteDescription(SessionDescription{Type: SDPTypeAnswer, SDP: s.Answer}); err != nil {
 				WebRTCPlugin.Error("Signal SetRemoteDescription", zap.Error(err))
 				return
 			}
 		}
+		WebRTCPlugin.Info(s.Type)
 	}
 }
--- a/go.mod
+++ b/go.mod
@ -3,9 +3,9 @@ module m7s.live/plugin/webrtc/v4
 go 1.18

 require (
-	github.com/pion/interceptor v0.1.11
+	github.com/pion/interceptor v0.1.12
 	github.com/pion/rtcp v1.2.10
-	github.com/pion/webrtc/v3 v3.1.49
+	github.com/pion/webrtc/v3 v3.1.56
 	go.uber.org/zap v1.23.0
 	m7s.live/engine/v4 v4.8.7
 )
@ -23,20 +23,20 @@ require (
 	github.com/marten-seemann/qtls-go1-18 v0.1.3 // indirect
 	github.com/marten-seemann/qtls-go1-19 v0.1.1 // indirect
 	github.com/onsi/ginkgo/v2 v2.2.0 // indirect
-	github.com/pion/datachannel v1.5.2 // indirect
-	github.com/pion/dtls/v2 v2.1.5 // indirect
-	github.com/pion/ice/v2 v2.2.12 // indirect
+	github.com/pion/datachannel v1.5.5 // indirect
+	github.com/pion/dtls/v2 v2.2.6 // indirect
+	github.com/pion/ice/v2 v2.3.1 // indirect
 	github.com/pion/logging v0.2.2 // indirect
-	github.com/pion/mdns v0.0.5 // indirect
+	github.com/pion/mdns v0.0.7 // indirect
 	github.com/pion/randutil v0.1.0 // indirect
 	github.com/pion/rtp v1.7.13 // indirect
-	github.com/pion/sctp v1.8.3 // indirect
+	github.com/pion/sctp v1.8.6 // indirect
 	github.com/pion/sdp/v3 v3.0.6 // indirect
-	github.com/pion/srtp/v2 v2.0.10 // indirect
-	github.com/pion/stun v0.3.5 // indirect
-	github.com/pion/transport v0.13.1 // indirect
-	github.com/pion/turn/v2 v2.0.8 // indirect
-	github.com/pion/udp v0.1.1 // indirect
+	github.com/pion/srtp/v2 v2.0.12 // indirect
+	github.com/pion/stun v0.4.0 // indirect
+	github.com/pion/transport/v2 v2.0.2 // indirect
+	github.com/pion/turn/v2 v2.1.0 // indirect
+	github.com/pion/udp/v2 v2.0.1 // indirect
 	github.com/power-devops/perfstat v0.0.0-20220216144756-c35f1ee13d7c // indirect
 	github.com/q191201771/naza v0.30.8 // indirect
 	github.com/shirou/gopsutil/v3 v3.22.10 // indirect
@ -45,7 +45,7 @@ require (
 	github.com/yusufpapurcu/wmi v1.2.2 // indirect
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
-	golang.org/x/crypto v0.3.0 // indirect
+	golang.org/x/crypto v0.6.0 // indirect
 	golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9 // indirect
 	golang.org/x/mod v0.7.0 // indirect
 	golang.org/x/net v0.7.0 // indirect
--- a/go.sum
+++ b/go.sum
@ -80,46 +80,45 @@ github.com/onsi/gomega v1.7.1/go.mod h1:XdKZgCCFLUoM/7CFJVPcG8C1xQ1AJ0vpAezJrB7J
 github.com/onsi/gomega v1.10.1/go.mod h1:iN09h71vgCQne3DLsj+A5owkum+a2tYe+TOCB1ybHNo=
 github.com/onsi/gomega v1.17.0/go.mod h1:HnhC7FXeEQY45zxNK3PPoIUhzk/80Xly9PcubAlGdZY=
 github.com/onsi/gomega v1.20.1 h1:PA/3qinGoukvymdIDV8pii6tiZgC8kbmJO6Z5+b002Q=
-github.com/pion/datachannel v1.5.2 h1:piB93s8LGmbECrpO84DnkIVWasRMk3IimbcXkTQLE6E=
-github.com/pion/datachannel v1.5.2/go.mod h1:FTGQWaHrdCwIJ1rw6xBIfZVkslikjShim5yr05XFuCQ=
-github.com/pion/dtls/v2 v2.1.5 h1:jlh2vtIyUBShchoTDqpCCqiYCyRFJ/lvf/gQ8TALs+c=
-github.com/pion/dtls/v2 v2.1.5/go.mod h1:BqCE7xPZbPSubGasRoDFJeTsyJtdD1FanJYL0JGheqY=
-github.com/pion/ice/v2 v2.2.12 h1:n3M3lUMKQM5IoofhJo73D3qVla+mJN2nVvbSPq32Nig=
-github.com/pion/ice/v2 v2.2.12/go.mod h1:z2KXVFyRkmjetRlaVRgjO9U3ShKwzhlUylvxKfHfd5A=
-github.com/pion/interceptor v0.1.11 h1:00U6OlqxA3FFB50HSg25J/8cWi7P6FbSzw4eFn24Bvs=
-github.com/pion/interceptor v0.1.11/go.mod h1:tbtKjZY14awXd7Bq0mmWvgtHB5MDaRN7HV3OZ/uy7s8=
+github.com/pion/datachannel v1.5.5 h1:10ef4kwdjije+M9d7Xm9im2Y3O6A6ccQb0zcqZcJew8=
+github.com/pion/datachannel v1.5.5/go.mod h1:iMz+lECmfdCMqFRhXhcA/219B0SQlbpoR2V118yimL0=
+github.com/pion/dtls/v2 v2.2.6 h1:yXMxKr0Skd+Ub6A8UqXTRLSywskx93ooMRHsQUtd+Z4=
+github.com/pion/dtls/v2 v2.2.6/go.mod h1:t8fWJCIquY5rlQZwA2yWxUS1+OCrAdXrhVKXB5oD/wY=
+github.com/pion/ice/v2 v2.3.1 h1:FQCmUfZe2Jpe7LYStVBOP6z1DiSzbIateih3TztgTjc=
+github.com/pion/ice/v2 v2.3.1/go.mod h1:aq2kc6MtYNcn4XmMhobAv6hTNJiHzvD0yXRz80+bnP8=
+github.com/pion/interceptor v0.1.12 h1:CslaNriCFUItiXS5o+hh5lpL0t0ytQkFnUcbbCs2Zq8=
+github.com/pion/interceptor v0.1.12/go.mod h1:bDtgAD9dRkBZpWHGKaoKb42FhDHTG2rX8Ii9LRALLVA=
 github.com/pion/logging v0.2.2 h1:M9+AIj/+pxNsDfAT64+MAVgJO0rsyLnoJKCqf//DoeY=
 github.com/pion/logging v0.2.2/go.mod h1:k0/tDVsRCX2Mb2ZEmTqNa7CWsQPc+YYCB7Q+5pahoms=
-github.com/pion/mdns v0.0.5 h1:Q2oj/JB3NqfzY9xGZ1fPzZzK7sDSD8rZPOvcIQ10BCw=
-github.com/pion/mdns v0.0.5/go.mod h1:UgssrvdD3mxpi8tMxAXbsppL3vJ4Jipw1mTCW+al01g=
+github.com/pion/mdns v0.0.7 h1:P0UB4Sr6xDWEox0kTVxF0LmQihtCbSAdW0H2nEgkA3U=
+github.com/pion/mdns v0.0.7/go.mod h1:4iP2UbeFhLI/vWju/bw6ZfwjJzk0z8DNValjGxR/dD8=
 github.com/pion/randutil v0.1.0 h1:CFG1UdESneORglEsnimhUjf33Rwjubwj6xfiOXBa3mA=
 github.com/pion/randutil v0.1.0/go.mod h1:XcJrSMMbbMRhASFVOlj/5hQial/Y8oH/HVo7TBZq+j8=
-github.com/pion/rtcp v1.2.9/go.mod h1:qVPhiCzAm4D/rxb6XzKeyZiQK69yJpbUDJSF7TgrqNo=
 github.com/pion/rtcp v1.2.10 h1:nkr3uj+8Sp97zyItdN60tE/S6vk4al5CPRR6Gejsdjc=
 github.com/pion/rtcp v1.2.10/go.mod h1:ztfEwXZNLGyF1oQDttz/ZKIBaeeg/oWbRYqzBM9TL1I=
 github.com/pion/rtp v1.6.2/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
 github.com/pion/rtp v1.7.13 h1:qcHwlmtiI50t1XivvoawdCGTP4Uiypzfrsap+bijcoA=
 github.com/pion/rtp v1.7.13/go.mod h1:bDb5n+BFZxXx0Ea7E5qe+klMuqiBrP+w8XSjiWtCUko=
-github.com/pion/sctp v1.8.0/go.mod h1:xFe9cLMZ5Vj6eOzpyiKjT9SwGM4KpK/8Jbw5//jc+0s=
-github.com/pion/sctp v1.8.3 h1:LWcciN2ptLkw9Ugp/Ks2E76fiWy7yk3Wm79D6oFbFNo=
-github.com/pion/sctp v1.8.3/go.mod h1:OHbDjdk7kg+L+7TJim9q/qGVefdEJohuA2SZyihccgI=
+github.com/pion/sctp v1.8.5/go.mod h1:SUFFfDpViyKejTAdwD1d/HQsCu+V/40cCs2nZIvC3s0=
+github.com/pion/sctp v1.8.6 h1:CUex11Vkt9YS++VhLf8b55O3VqKrWL6W3SDwX4jAqsI=
+github.com/pion/sctp v1.8.6/go.mod h1:SUFFfDpViyKejTAdwD1d/HQsCu+V/40cCs2nZIvC3s0=
 github.com/pion/sdp/v3 v3.0.6 h1:WuDLhtuFUUVpTfus9ILC4HRyHsW6TdugjEX/QY9OiUw=
 github.com/pion/sdp/v3 v3.0.6/go.mod h1:iiFWFpQO8Fy3S5ldclBkpXqmWy02ns78NOKoLLL0YQw=
-github.com/pion/srtp/v2 v2.0.10 h1:b8ZvEuI+mrL8hbr/f1YiJFB34UMrOac3R3N1yq2UN0w=
-github.com/pion/srtp/v2 v2.0.10/go.mod h1:XEeSWaK9PfuMs7zxXyiN252AHPbH12NX5q/CFDWtUuA=
-github.com/pion/stun v0.3.5 h1:uLUCBCkQby4S1cf6CGuR9QrVOKcvUwFeemaC865QHDg=
-github.com/pion/stun v0.3.5/go.mod h1:gDMim+47EeEtfWogA37n6qXZS88L5V6LqFcf+DZA2UA=
-github.com/pion/transport v0.12.2/go.mod h1:N3+vZQD9HlDP5GWkZ85LohxNsDcNgofQmyL6ojX5d8Q=
-github.com/pion/transport v0.12.3/go.mod h1:OViWW9SP2peE/HbwBvARicmAVnesphkNkCVZIWJ6q9A=
-github.com/pion/transport v0.13.0/go.mod h1:yxm9uXpK9bpBBWkITk13cLo1y5/ur5VQpG22ny6EP7g=
-github.com/pion/transport v0.13.1 h1:/UH5yLeQtwm2VZIPjxwnNFxjS4DFhyLfS4GlfuKUzfA=
-github.com/pion/transport v0.13.1/go.mod h1:EBxbqzyv+ZrmDb82XswEE0BjfQFtuw1Nu6sjnjWCsGg=
-github.com/pion/turn/v2 v2.0.8 h1:KEstL92OUN3k5k8qxsXHpr7WWfrdp7iJZHx99ud8muw=
-github.com/pion/turn/v2 v2.0.8/go.mod h1:+y7xl719J8bAEVpSXBXvTxStjJv3hbz9YFflvkpcGPw=
-github.com/pion/udp v0.1.1 h1:8UAPvyqmsxK8oOjloDk4wUt63TzFe9WEJkg5lChlj7o=
-github.com/pion/udp v0.1.1/go.mod h1:6AFo+CMdKQm7UiA0eUPA8/eVCTx8jBIITLZHc9DWX5M=
-github.com/pion/webrtc/v3 v3.1.49 h1:rbsNGxK9jMYts+xE6zYAJMUQHnGwmk/JYze8yttW+to=
-github.com/pion/webrtc/v3 v3.1.49/go.mod h1:kHf/o47QW4No1rgpsFux/h7lUhtUnwFnSFDZOXeLapw=
+github.com/pion/srtp/v2 v2.0.12 h1:WrmiVCubGMOAObBU1vwWjG0H3VSyQHawKeer2PVA5rY=
+github.com/pion/srtp/v2 v2.0.12/go.mod h1:C3Ep44hlOo2qEYaq4ddsmK5dL63eLehXFbHaZ9F5V9Y=
+github.com/pion/stun v0.4.0 h1:vgRrbBE2htWHy7l3Zsxckk7rkjnjOsSM7PHZnBwo8rk=
+github.com/pion/stun v0.4.0/go.mod h1:QPsh1/SbXASntw3zkkrIk3ZJVKz4saBY2G7S10P3wCw=
+github.com/pion/transport v0.14.1 h1:XSM6olwW+o8J4SCmOBb/BpwZypkHeyM0PGFCxNQBr40=
+github.com/pion/transport v0.14.1/go.mod h1:4tGmbk00NeYA3rUa9+n+dzCCoKkcy3YlYb99Jn2fNnI=
+github.com/pion/transport/v2 v2.0.0/go.mod h1:HS2MEBJTwD+1ZI2eSXSvHJx/HnzQqRy2/LXxt6eVMHc=
+github.com/pion/transport/v2 v2.0.2 h1:St+8o+1PEzPT51O9bv+tH/KYYLMNR5Vwm5Z3Qkjsywg=
+github.com/pion/transport/v2 v2.0.2/go.mod h1:vrz6bUbFr/cjdwbnxq8OdDDzHf7JJfGsIRkxfpZoTA0=
+github.com/pion/turn/v2 v2.1.0 h1:5wGHSgGhJhP/RpabkUb/T9PdsAjkGLS6toYz5HNzoSI=
+github.com/pion/turn/v2 v2.1.0/go.mod h1:yrT5XbXSGX1VFSF31A3c1kCNB5bBZgk/uu5LET162qs=
+github.com/pion/udp/v2 v2.0.1 h1:xP0z6WNux1zWEjhC7onRA3EwwSliXqu1ElUZAQhUP54=
+github.com/pion/udp/v2 v2.0.1/go.mod h1:B7uvTMP00lzWdyMr/1PVZXtV3wpPIxBRd4Wl6AksXn8=
+github.com/pion/webrtc/v3 v3.1.56 h1:ScaiqKQN3liQwT+kJwOBaYP6TwSfixzdUnZmzHAo0a0=
+github.com/pion/webrtc/v3 v3.1.56/go.mod h1:7VhbA6ihqJlz6R/INHjyh1b8HpiV9Ct4UQvE1OB/xoM=
 github.com/pixelbender/go-sdp v1.1.0/go.mod h1:6IBlz9+BrUHoFTea7gcp4S54khtOhjCW/nVDLhmZBAs=
 github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
 github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
@ -169,10 +168,9 @@ golang.org/x/crypto v0.0.0-20191011191535-87dc89f01550/go.mod h1:yigFU9vqHzYiE8U
 golang.org/x/crypto v0.0.0-20200622213623-75b288015ac9/go.mod h1:LzIPMQfyMNhhGPhUkYOs5KpL4U8rLKemX1yGLhDgUto=
 golang.org/x/crypto v0.0.0-20201221181555-eec23a3978ad/go.mod h1:jdWPYTVW3xRLrWPugEBEK3UY2ZEsg3UU495nc5E+M+I=
 golang.org/x/crypto v0.0.0-20210921155107-089bfa567519/go.mod h1:GvvjBRRGRdwPK5ydBHafDWAxML/pGHZbMvKqRZ5+Abc=
-golang.org/x/crypto v0.0.0-20220427172511-eb4f295cb31f/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.0.0-20221010152910-d6f0a8c073c2/go.mod h1:IxCIyHEi3zRg3s0A5j5BB6A9Jmi73HwBIUl50j+osU4=
-golang.org/x/crypto v0.3.0 h1:a06MkbcxBrEFc0w0QIZWXrH/9cCX6KJyWbBOIwAn+7A=
-golang.org/x/crypto v0.3.0/go.mod h1:hebNnKkNXi2UzZN1eVRvBB7co0a+JxK6XbPiWVs/3J4=
+golang.org/x/crypto v0.5.0/go.mod h1:NK/OQwhpMQP3MwtdjgLlYHnH9ebylxKWv3e0fK+mkQU=
+golang.org/x/crypto v0.6.0 h1:qfktjS5LUO+fFKeJXZ+ikTRijMmljikvG68fpMMruSc=
+golang.org/x/crypto v0.6.0/go.mod h1:OFC/31mSvZgRz0V1QTNCzfAI1aIRzbiufJtkMIlEp58=
 golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9 h1:yZNXmy+j/JpX19vZkVktWqAo7Gny4PBWYYK3zskGpx4=
 golang.org/x/exp v0.0.0-20221126150942-6ab00d035af9/go.mod h1:CxIveKay+FTh1D0yPZemJVgC/95VzuuOLq5Qi4xnoYc=
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
@ -185,17 +183,13 @@ golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
 golang.org/x/net v0.0.0-20200520004742-59133d7f0dd7/go.mod h1:qpuaurCH72eLCgpAm/N6yyVIVM9cpaDIP3A8BGJEC5A=
 golang.org/x/net v0.0.0-20201021035429-f5854403a974/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20201201195509-5d6afe98e0b7/go.mod h1:sp8m0HH+o8qH0wwXwYZr8TS3Oi6o0r6Gce1SSxlDquU=
-golang.org/x/net v0.0.0-20210119194325-5f4716e94777/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210226172049-e18ecbb05110/go.mod h1:m0MpNAwzfU5UDzcl9v0D8zg8gWTRqZa9RBIspLL5mdg=
 golang.org/x/net v0.0.0-20210405180319-a5a99cb37ef4/go.mod h1:p54w0d4576C0XHj96bSt6lcn1PtDYWL6XObtHCRCNQM=
 golang.org/x/net v0.0.0-20210428140749-89ef3d95e781/go.mod h1:OJAsFXCWl8Ukc7SiCT/9KSuxbyM7479/AVlXFRxuMCk=
-golang.org/x/net v0.0.0-20211112202133-69e39bad7dc2/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20211201190559-0a0e4e1bb54c/go.mod h1:9nx3DQGgdP8bBQD5qxJ1jj9UTztislL4KSBs9R2vV5Y=
-golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su4UZFWDARRcnwPjda9FqA0JpMk=
-golang.org/x/net v0.0.0-20220531201128-c960675eff93/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.0.0-20220722155237-a158d28d115b/go.mod h1:XRhObCWvk6IyKnWLug+ECip1KBveYUHfp+8e9klMJ9c=
 golang.org/x/net v0.1.0/go.mod h1:Cx3nUiGt4eDBEyega/BKRp+/AlGL8hYe7U9odMt2Cco=
+golang.org/x/net v0.5.0/go.mod h1:DivGGAXEgPSlEBzxGzZI+ZLohi+xUj054jfeKui00ws=
+golang.org/x/net v0.6.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/net v0.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
 golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@ -223,26 +217,28 @@ golang.org/x/sys v0.0.0-20210330210617-4fbd30eecc44/go.mod h1:h1NjWce9XRLGQEsW7w
 golang.org/x/sys v0.0.0-20210423082822-04245dca01da/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20210510120138-977fb7262007/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20210615035016-665e8c7367d1/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20211216021012-1d35b9e2eb4e/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220128215802-99c3d69c2c27/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220520151302-bc2c85ada10a/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220608164250-635b8c9b7f68/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
-golang.org/x/sys v0.0.0-20220622161953-175b2fd9d664/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.0.0-20220722155257-8c9f86f7a55f/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.1.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.2.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.4.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/sys v0.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
 golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
 golang.org/x/term v0.0.0-20201117132131-f5c789dd3221/go.mod h1:Nr5EML6q2oocZ2LXRh80K7BxOlk5/8JxuGnuhpl+muw=
 golang.org/x/term v0.0.0-20201126162022-7de9c90e9dd1/go.mod h1:bj7SfCRtBDWHUb9snDiAeCFNEtKQo2Wmx5Cou7ajbmo=
 golang.org/x/term v0.0.0-20210927222741-03fcf44c2211/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
 golang.org/x/term v0.1.0/go.mod h1:jbD1KX2456YbFQfuXm/mYQcufACuNUgVhRMnK/tPxf8=
+golang.org/x/term v0.4.0/go.mod h1:9P2UbLfCdcvo3p/nzKvsmas4TnlujnuoV9hGgYzW1lQ=
+golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.3/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.6/go.mod h1:5Zoc/QRtKVWzQhOtBMvqHzDpF6irO9z98xDceosuGiQ=
 golang.org/x/text v0.3.7/go.mod h1:u+2+/6zg+i71rQMx5EYifcz6MCKuco9NR6JIITiCfzQ=
 golang.org/x/text v0.4.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.6.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/text v0.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
+golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
 golang.org/x/tools v0.0.0-20180917221912-90fa682c2a6e/go.mod h1:n7NCudcB/nEzxVGmLbDWY5pfWTLqBcC2KZ6jyYvM4mQ=
 golang.org/x/tools v0.0.0-20191119224855-298f0cb1881e/go.mod h1:b+2E5dAYhXwXZwtnZ6UAqBI28+e2cm9otk0dWdXHAEo=
 golang.org/x/tools v0.0.0-20201224043029-2b0845dc783e/go.mod h1:emZCQorbCU4vsT4fOWvOPXz4eW1wZW4PmDk9uLelYpA=
--- a/main.go
+++ b/main.go
@ -144,23 +144,53 @@ func (conf *WebRTCConfig) Push_(w http.ResponseWriter, r *http.Request) {
 		http.Error(w, err.Error(), http.StatusInternalServerError)
 		return
 	}
+	// puber.SetIO(puber.PeerConnection)
 	puber.OnICECandidate(func(ice *ICECandidate) {
 		if ice != nil {
 			puber.Info(ice.ToJSON().Candidate)
 		}
 	})
-	if _, err = puber.AddTransceiverFromKind(RTPCodecTypeVideo); err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
-	if _, err = puber.AddTransceiverFromKind(RTPCodecTypeAudio); err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-		return
-	}
+	puber.OnDataChannel(func(d *DataChannel) {
+		puber.Info("OnDataChannel", zap.String("label", d.Label()))
+		d.OnMessage(func(msg DataChannelMessage) {
+			puber.SDP = string(msg.Data[1:])
+			puber.Debug("dc message", zap.String("sdp", puber.SDP))
+			if err := puber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: puber.SDP}); err != nil {
+				return
+			}
+			if answer, err := puber.GetAnswer(); err == nil {
+				d.SendText(answer)
+			} else {
+				return
+			}
+			switch msg.Data[0] {
+			case '0':
+				puber.Stop()
+			case '1':
+				
+			}
+		})
+	})
+	// if _, err = puber.AddTransceiverFromKind(RTPCodecTypeVideo); err != nil {
+	// 	http.Error(w, err.Error(), http.StatusInternalServerError)
+	// 	return
+	// }
+	// if _, err = puber.AddTransceiverFromKind(RTPCodecTypeAudio); err != nil {
+	// 	http.Error(w, err.Error(), http.StatusInternalServerError)
+	// 	return
+	// }
 	if err = WebRTCPlugin.Publish(streamPath, &puber); err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
 	}
+	puber.OnConnectionStateChange(func(state PeerConnectionState) {
+		switch state {
+		case PeerConnectionStateConnected:
+
+		case PeerConnectionStateDisconnected, PeerConnectionStateFailed:
+			puber.Stop()
+		}
+	})
 	if err := puber.SetRemoteDescription(SessionDescription{Type: SDPTypeOffer, SDP: puber.SDP}); err != nil {
 		http.Error(w, err.Error(), http.StatusBadRequest)
 		return
--- a/publish.html
+++ b/publish.html
@ -2,54 +2,83 @@
 <html lang="en">

 <head>
-    <meta charset="UTF-8">
-    <meta http-equiv="X-UA-Compatible" content="IE=edge">
-    <meta name="viewport" content="width=device-width, initial-scale=1.0">
-    <title>测试WebRTC推流</title>
+  <meta charset="UTF-8">
+  <meta http-equiv="X-UA-Compatible" content="IE=edge">
+  <meta name="viewport" content="width=device-width, initial-scale=1.0">
+  <title>测试WebRTC推流</title>
 </head>

 <body>
-    <video id="video" width="640" height="480" autoplay muted />
+  <video id="video" width="640" height="480" autoplay muted>
+  </video>
+  <!-- <button id="sw" onclick="action()" type="button" style="width:100px;height:30px;display: block;">unpublish</button> -->
 </body>
 <script>
-    (async () => {
-        const mediaStream = await navigator.mediaDevices.getUserMedia({
-            video: true,
-            audio: true,
-        });
-        document.getElementById('video').srcObject = mediaStream;
-        const pc = new RTCPeerConnection();
-        pc.addTransceiver('video', { direction: 'sendonly' });
-        pc.addTransceiver('audio', { direction: 'sendonly' });
-        pc.oniceconnectionstatechange = () => {
-            console.log('oniceconnectionstatechange', pc.iceConnectionState);
-        };
-        pc.onicecandidate = (e) => {
-            console.log('onicecandidate', e.candidate);
-        };
-        mediaStream.getTracks().forEach((t) => {
-            pc.addTrack(t, mediaStream);
-        });
-        const offer = await pc.createOffer();
-        await pc.setLocalDescription(offer);
-        const result = await fetch(
-            `/webrtc/push/live/webrtc`,
-            {
-                method: 'POST',
-                mode: 'cors',
-                cache: 'no-cache',
-                credentials: 'include',
-                redirect: 'follow',
-                referrerPolicy: 'no-referrer',
-                headers: { 'Content-Type': 'application/sdp' },
-                body: offer.sdp,
-            },
-        );
-        const remoteSdp = await result.text();
-        await pc.setRemoteDescription(
-            new RTCSessionDescription({ type: 'answer', sdp: remoteSdp }),
-        );
-    })()
+  let action = () => { console.log('action not set'); };
+  (async () => {
+    const mediaStream = await navigator.mediaDevices.getUserMedia({
+      video: true,
+      audio: true,
+    });
+    document.getElementById('video').srcObject = mediaStream;
+    const pc = new RTCPeerConnection();
+
+    pc.oniceconnectionstatechange = () => {
+      console.log('oniceconnectionstatechange', pc.iceConnectionState);
+    };
+    pc.onicecandidate = (e) => {
+      console.log('onicecandidate', e.candidate);
+    };
+    // mediaStream.getTracks().forEach((t) => {
+    //     pc.addTrack(t, mediaStream);
+    // });
+    let videoTransceiver = pc.addTransceiver(mediaStream.getVideoTracks()[0], { direction: 'sendonly' });
+    let audioTransceiver = pc.addTransceiver(mediaStream.getAudioTracks()[0], { direction: 'sendonly' });
+    const dc = pc.createDataChannel('sdp');
+    const offer = await pc.createOffer();
+    await pc.setLocalDescription(offer);
+    const result = await fetch(
+      `/webrtc/push/live/webrtc`,
+      {
+        method: 'POST',
+        mode: 'cors',
+        cache: 'no-cache',
+        credentials: 'include',
+        redirect: 'follow',
+        referrerPolicy: 'no-referrer',
+        headers: { 'Content-Type': 'application/sdp' },
+        body: offer.sdp,
+      },
+    );
+    const remoteSdp = await result.text();
+    await pc.setRemoteDescription(
+      new RTCSessionDescription({ type: 'answer', sdp: remoteSdp }),
+    );
+    dc.onmessage = async (e) => {
+      await pc.setRemoteDescription(
+        new RTCSessionDescription({ type: 'answer', sdp: e.data }),
+      );
+    };
+    const publish = async () => {
+      videoTransceiver.direction = 'sendonly';
+      audioTransceiver.direction = 'sendonly';
+      const offer = await pc.createOffer();
+      await pc.setLocalDescription(offer);
+      dc.send('1' + offer.sdp);
+      action = unpublish;
+      document.getElementById('sw').innerText = 'unpublish';
+    };
+    const unpublish = async () => {
+      videoTransceiver.direction = 'inactive';
+      audioTransceiver.direction = 'inactive';
+      const offer = await pc.createOffer();
+      await pc.setLocalDescription(offer);
+      dc.send('0' + offer.sdp);
+      action = publish;
+      document.getElementById('sw').innerText = 'publish';
+    };
+    action = unpublish;
+  })()
 </script>

 </html>
--- a/publisher.go
+++ b/publisher.go
@ -6,6 +6,7 @@ import (

 	"github.com/pion/rtcp"
 	. "github.com/pion/webrtc/v3"
+	"go.uber.org/zap"
 	. "m7s.live/engine/v4"
 	. "m7s.live/engine/v4/track"
 )
@ -16,63 +17,68 @@ type WebRTCPublisher struct {
 }

 func (puber *WebRTCPublisher) OnEvent(event any) {
-	switch v := event.(type) {
+	switch event.(type) {
 	case IPublisher:
-		puber.OnICEConnectionStateChange(func(connectionState ICEConnectionState) {
-			puber.Info("Connection State has changed:" + connectionState.String())
-			switch connectionState {
-			case ICEConnectionStateDisconnected, ICEConnectionStateFailed:
-				puber.Stop()
+		puber.OnTrack(puber.onTrack)
+	}
+	puber.Publisher.OnEvent(event)
+}
+
+func (puber *WebRTCPublisher) onTrack(track *TrackRemote, receiver *RTPReceiver) {
+	puber.Info("onTrack", zap.String("kind", track.Kind().String()), zap.Uint8("payloadType", uint8(track.Codec().PayloadType)))
+	if codec := track.Codec(); track.Kind() == RTPCodecTypeAudio {
+		if puber.AudioTrack == nil {
+			switch codec.PayloadType {
+			case 8:
+				puber.AudioTrack = NewG711(puber.Stream, true)
+			case 0:
+				puber.AudioTrack = NewG711(puber.Stream, false)
+			default:
+				puber.AudioTrack = nil
+				return
+			}
+		}
+		for {
+			rtpItem := puber.AudioTrack.GetRTPFromPool()
+			if i, _, err := track.Read(rtpItem.Value.Raw[:1460]); err == nil {
+				rtpItem.Value.Unmarshal(rtpItem.Value.Raw[:i])
+				puber.AudioTrack.WriteRTP(rtpItem)
+			} else {
+				puber.Info("track stop", zap.String("kind", track.Kind().String()), zap.Error(err))
+				rtpItem.Recycle()
+				return
 			}
-		})
-		puber.OnTrack(func(track *TrackRemote, receiver *RTPReceiver) {
-			if codec := track.Codec(); track.Kind() == RTPCodecTypeAudio {
-				if puber.Equal(v) || puber.AudioTrack == nil {
-					switch codec.PayloadType {
-					case 8:
-						puber.AudioTrack = NewG711(puber.Stream, true)
-					case 0:
-						puber.AudioTrack = NewG711(puber.Stream, false)
-					default:
-						puber.AudioTrack = nil
-						return
-					}
-				}
-				for {
-					b := make([]byte, 1460)
-					if i, _, err := track.Read(b); err == nil {
-						puber.AudioTrack.WriteRTP(b[:i])
-					} else {
-						return
-					}
-				}
+		}
+	} else {
+		go puber.writeRTCP(track)
+		if puber.VideoTrack == nil {
+			puber.VideoTrack = NewH264(puber.Stream, byte(codec.PayloadType))
+		}
+		for {
+			rtpItem := puber.VideoTrack.GetRTPFromPool()
+			if i, _, err := track.Read(rtpItem.Value.Raw[:1460]); err == nil {
+				rtpItem.Value.Unmarshal(rtpItem.Value.Raw[:i])
+				puber.VideoTrack.WriteRTP(rtpItem)
 			} else {
-				go func() {
-					ticker := time.NewTicker(webrtcConfig.PLI)
-					for {
-						select {
-						case <-ticker.C:
-							if rtcpErr := puber.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}}); rtcpErr != nil {
-								fmt.Println(rtcpErr)
-							}
-						case <-puber.Done():
-							return
-						}
-					}
-				}()
-				if puber.Equal(v) {
-					puber.VideoTrack = NewH264(puber.Stream, byte(codec.PayloadType))
-				}
-				for {
-					b := make([]byte, 1460)
-					if i, _, err := track.Read(b); err == nil {
-						puber.VideoTrack.WriteRTP(b[:i])
-					} else {
-						return
-					}
-				}
+				puber.Info("track stop", zap.String("kind", track.Kind().String()), zap.Error(err))
+				rtpItem.Recycle()
+				return
 			}
-		})
+		}
+	}
+}
+
+func (puber *WebRTCPublisher) writeRTCP(track *TrackRemote) {
+	ticker := time.NewTicker(webrtcConfig.PLI)
+	defer ticker.Stop()
+	for {
+		select {
+		case <-ticker.C:
+			if rtcpErr := puber.WriteRTCP([]rtcp.Packet{&rtcp.PictureLossIndication{MediaSSRC: uint32(track.SSRC())}}); rtcpErr != nil {
+				fmt.Println(rtcpErr)
+			}
+		case <-puber.Done():
+			return
+		}
 	}
-	puber.Publisher.OnEvent(event)
 }
--- a/subscriber.go
+++ b/subscriber.go
@ -86,7 +86,7 @@ func (suber *WebRTCSubscriber) OnEvent(event any) {
 		}
 	case VideoRTP:
 		if suber.videoTrack != nil {
-			suber.videoTrack.WriteRTP(&v.Packet)
+			suber.videoTrack.WriteRTP(v.Packet)
 		} else if suber.DC != nil {
 			frame := suber.VideoReader.Frame
 			dataSize := uint32(frame.AVCC.ByteLength)
@ -104,7 +104,7 @@ func (suber *WebRTCSubscriber) OnEvent(event any) {
 			}
 		}
 	case AudioRTP:
-		suber.audioTrack.WriteRTP(&v.Packet)
+		suber.audioTrack.WriteRTP(v.Packet)
 	case ISubscriber:
 		suber.OnConnectionStateChange(func(pcs PeerConnectionState) {
 			suber.Info("Connection State has changed:" + pcs.String())