增加调用发布者的Dispose方法

amf.go

@@ -8,8 +8,7 @@ import (
 	"log"
 	"reflect"
 
-	"github.com/Monibuca/engine/v2/pool"
+	"github.com/Monibuca/utils/v3"
-	"github.com/Monibuca/engine/v2/util"
 )
 
 // Action Message Format -- AMF 0
@@ -110,14 +109,14 @@ func newAMFDecoder(b []byte) *AMF {
 }
 func (amf *AMF) readSize() (int, error) {
 	b, err := readBytes(amf.Buffer, 4)
-	size := int(util.BigEndian.Uint32(b))
+	size := int(utils.BigEndian.Uint32(b))
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	return size, err
 }
 func (amf *AMF) readSize16() (int, error) {
 	b, err := readBytes(amf.Buffer, 2)
-	size := int(util.BigEndian.Uint16(b))
+	size := int(utils.BigEndian.Uint16(b))
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	return size, err
 }
 func (amf *AMF) readObjects() (obj []AMFObject, err error) {
@@ -220,7 +219,7 @@ func (amf *AMF) encodeObject(t AMFObjects) (err error) {
 				return
 			}
 		case float64, uint, float32, int, int16, int32, int64, uint16, uint32, uint64, uint8, int8:
-			if err = amf.writeObjectNumber(k, util.ToFloat64(vvv)); err != nil {
+			if err = amf.writeObjectNumber(k, utils.ToFloat64(vvv)); err != nil {
 				return
 			}
 		case bool:
@@ -236,10 +235,10 @@ func (amf *AMF) readDate() (t uint64, err error) {
 	_, err = amf.ReadByte() // 取出第一个字节 8 Bit == 1 Byte. buf - 1.
 	var b []byte
 	b, err = readBytes(amf.Buffer, 8) // 在取出8个字节,并且读到b中. buf - 8
-	t = util.BigEndian.Uint64(b)
+	t = utils.BigEndian.Uint64(b)
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	b, err = readBytes(amf.Buffer, 2)
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	return t, err
 }
 
@@ -290,7 +289,7 @@ func (amf *AMF) readString1() (str string, err error) {
 	var b []byte
 	b, err = readBytes(amf.Buffer, size) // 读取全部数据,读取长度为l,因为这两个字节(l变量)保存的是数据长度
 	str = string(b)
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	return
 }
 
@@ -301,7 +300,7 @@ func (amf *AMF) readLongString() (str string, err error) {
 	var b []byte
 	b, err = readBytes(amf.Buffer, size) // 读取全部数据,读取长度为l,因为这两个字节(l变量)保存的是数据长度
 	str = string(b)
-	pool.RecycleSlice(b)
+	utils.RecycleSlice(b)
 	return
 }
 
@@ -346,16 +345,16 @@ func (amf *AMF) readObject() (m AMFObjects, err error) {
 }
 
 func readBytes(buf *bytes.Buffer, length int) (b []byte, err error) {
-	b = pool.GetSlice(length)
+	b = utils.GetSlice(length)
 	if i, _ := buf.Read(b); length != i {
 		err = errors.New(fmt.Sprintf("not enough bytes,%v/%v", buf.Len(), length))
 	}
 	return
 }
 func (amf *AMF) writeSize16(l int) (err error) {
-	b := pool.GetSlice(2)
+	b := utils.GetSlice(2)
-	defer pool.RecycleSlice(b)
+	defer utils.RecycleSlice(b)
-	util.BigEndian.PutUint16(b, uint16(l))
+	utils.BigEndian.PutUint16(b, uint16(l))
 	_, err = amf.Write(b)
 	return
 }

go.mod

@@ -3,11 +3,7 @@ module github.com/Monibuca/plugin-rtmp/v3
 go 1.13
 
 require (
-	github.com/Monibuca/engine/v3 v3.0.1
+	github.com/Monibuca/engine/v3 v3.0.0-alpha2
-	github.com/Monibuca/utils/v3 v3.0.0-alpha2
+	github.com/Monibuca/utils/v3 v3.0.0-alpha3
 	github.com/logrusorgru/aurora v2.0.3+incompatible
 )
-
-replace github.com/Monibuca/engine/v3 => ../engine
-
-replace github.com/Monibuca/utils/v3 v3.0.0-alpha2 => ../utils

go.sum

@@ -1,46 +1,23 @@
 github.com/BurntSushi/toml v0.3.1 h1:WXkYYl6Yr3qBf1K79EBnL4mak0OimBfB0XUf9Vl28OQ=
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/Monibuca/engine v1.2.2 h1:hNjsrZpOmui8lYhgCJ5ltJU8g/k0Rrdysx2tHNGGnbI=
+github.com/Monibuca/engine/v3 v3.0.0-alpha2 h1:11rWc6Qnp3kuVwGh8XDXqRivUSWMyv+Au6wZDgNHPSc=
-github.com/Monibuca/engine v1.2.2/go.mod h1:WbDkXENLjcPjyjCR1Mix1GA+uAlwORkv/+8aMVrDX2g=
+github.com/Monibuca/engine/v3 v3.0.0-alpha2/go.mod h1:K5FB3wk+iS/nPc+NS4XwObYQV4gtF6klEwDxaVM0BsQ=
-github.com/Monibuca/engine/v2 v2.0.0-alpha1 h1:RT+rKOJC1MdFNZp6gADGdzGLxEhYN35MKEAx7FwHtps=
+github.com/Monibuca/utils/v3 v3.0.0-alpha3 h1:n4Sq7mS1Iz8oBj2BcV4sXgKbZgix0fFLvjAfXYoiXl0=
-github.com/Monibuca/engine/v2 v2.0.0-alpha1/go.mod h1:34EYjjV15G6myuHOKaJkO7y5tJ1Arq/NfC9Weacr2mc=
+github.com/Monibuca/utils/v3 v3.0.0-alpha3/go.mod h1:3xYmhQbgAZBHLyIMteUCd1va+1z/xnd72B585mCaT3c=
-github.com/Monibuca/engine/v2 v2.0.0 h1:8FjaScrtN8QdbcxO9zZYABMC0Re3I1O1T4p94zAXYb0=
-github.com/Monibuca/engine/v2 v2.0.0/go.mod h1:34EYjjV15G6myuHOKaJkO7y5tJ1Arq/NfC9Weacr2mc=
-github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d h1:G0m3OIz70MZUWq3EgK3CesDbo8upS2Vm9/P3FtgI+Jk=
-github.com/StackExchange/wmi v0.0.0-20190523213315-cbe66965904d/go.mod h1:3eOhrUMpNV+6aFIbp5/iudMxNCF27Vw2OZgy4xEx0Fg=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478 h1:Db9StoJ6RZN3YttC0Pm0I4Y5izITRYch3RMbT59BYN0=
 github.com/funny/slab v0.0.0-20180511031532-b1fad5e5d478/go.mod h1:0j1+svBH8ABEIPdUP0AIg4qedsybnXGJBakCEw8cfoo=
 github.com/funny/utest v0.0.0-20161029064919-43870a374500 h1:Z0r1CZnoIWFB/Uiwh1BU5FYmuFe6L5NPi6XWQEmsTRg=
 github.com/funny/utest v0.0.0-20161029064919-43870a374500/go.mod h1:mUn39tBov9jKnTWV1RlOYoNzxdBFHiSzXWdY1FoNGGg=
-github.com/go-ole/go-ole v1.2.4 h1:nNBDSCOigTSiarFpYE9J/KtEA1IOW4CNeqT9TQDqCxI=
-github.com/go-ole/go-ole v1.2.4/go.mod h1:XCwSNxSkXRo4vlyPy93sltvi/qJq0jqQhjqQNIwKuxM=
-github.com/logrusorgru/aurora v0.0.0-20200102142835-e9ef32dff381 h1:bqDmpDG49ZRnB5PcgP0RXtQvnMSgIF14M7CBd2shtXs=
-github.com/logrusorgru/aurora v0.0.0-20200102142835-e9ef32dff381/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
 github.com/logrusorgru/aurora v2.0.3+incompatible h1:tOpm7WcpBTn4fjmVfgpQq0EfczGlG91VSDkswnjF5A8=
 github.com/logrusorgru/aurora v2.0.3+incompatible/go.mod h1:7rIyQOR62GCctdiQpZ/zOJlFyk6y+94wXzv6RNZgaR4=
-github.com/mattn/go-colorable v0.1.6 h1:6Su7aK7lXmJ/U79bYtBjLNaha4Fs1Rg9plHpcH+vvnE=
-github.com/mattn/go-colorable v0.1.6/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-colorable v0.1.8 h1:c1ghPdyEDarC70ftn0y+A/Ee++9zz8ljHG1b13eJ0s8=
 github.com/mattn/go-colorable v0.1.8/go.mod h1:u6P/XSegPjTcexA+o6vUJrdnUu04hMope9wVRipJSqc=
 github.com/mattn/go-isatty v0.0.12 h1:wuysRhFDzyxgEmMf5xjvJ2M9dZoWAXNNr5LSBS7uHXY=
 github.com/mattn/go-isatty v0.0.12/go.mod h1:cbi8OIDigv2wuxKPP5vlRcQ1OAZbq2CE4Kysco4FUpU=
 github.com/pkg/errors v0.9.1 h1:FEBLx1zS214owpjy7qsBeixbURkuhQAwrK5UwLGTwt4=
 github.com/pkg/errors v0.9.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/shirou/gopsutil v2.20.1+incompatible h1:oIq9Cq4i84Hk8uQAUOG3eNdI/29hBawGrD5YRl6JRDY=
-github.com/shirou/gopsutil v2.20.1+incompatible/go.mod h1:5b4v6he4MtMOwMlS0TUMTu2PcXUg8+E1lC7eC3UO/RA=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.5.1 h1:nOGnQDM7FYENwehXlg/kFVnos3rEvtKTjRvOWSzb6H4=
-github.com/stretchr/testify v1.5.1/go.mod h1:5W2xD1RspED5o8YsWQXVCued0rvSQ+mT+I5cxcmMvtA=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a h1:DcqTD9SDLc+1P/r1EmRBwnVsrOwW+kk2vWf9n+1sGhs=
 golang.org/x/sync v0.0.0-20201207232520-09787c993a3a/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sys v0.0.0-20200116001909-b77594299b42/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae h1:/WDfKMnPU+m5M4xB+6x4kaepxRw6jWvR5iDRdvjHgy8=
 golang.org/x/sys v0.0.0-20200223170610-d5e6a3e2c0ae/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405 h1:yhCVgyC4o1eVCa2tZl7eS0r+SDo693bJlVdllGtEeKM=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=

netStream.go

@@ -55,6 +55,9 @@ func processRtmp(conn net.Conn) {
 	defer func() {
 		conn.Close()
 		if stream != nil {
+			if stream.Publisher != nil {
+				stream.Publisher.Dispose()
+			}
 			stream.Close()
 		}
 		for _, s := range streams {