feat: change to use ps plugin

feat: add a new way to config port
ChannelEx
2025-12-24 13:27:57 +08:00 · 2023-05-14 11:12:25 +08:00 · 2023-05-04 09:57:20 +08:00 · 2023-04-09 11:08:23 +08:00 · 2023-04-05 22:25:53 +08:00 · 2023-04-04 20:20:27 +08:00
12 changed files with 256 additions and 584 deletions
--- a/README.md
+++ b/README.md
@@ -18,59 +18,33 @@ _ "m7s.live/plugin/gb28181/v4"

 ```yaml
 gb28181:
-  autoinvite:     true
+  autoinvite:     true #表示自动发起invite，当Server（SIP）接收到设备信息时，立即向设备发送invite命令获取流
  position:
    autosubposition: false #是否自动订阅定位
    expires: 3600s #订阅周期(单位：秒)，默认3600
    interval: 6s #订阅间隔（单位：秒），默认6
-  prefetchrecord: false
-  udpcachesize:   0
+  prefetchrecord: false 
+  udpcachesize:   0 #表示UDP缓存大小，默认为0，不开启。仅当TCP关闭，切缓存大于0时才开启
  sipnetwork:     udp
-  sipip:          ""
+  sipip:          "" #sip服务器地址 默认 自动适配设备网段
  sipport:        5060
  serial:         "34020000002000000001"
  realm:          "3402000000"
  username:       ""
  password:       ""
  
-  registervalidity:  60s
+  registervalidity:  60s #注册有效期
  
-  mediaip:          ""
-  mediaport:        58200
-  mediaidletimeout: 30
-  medianetwork:     udp
-  mediaportmin:     0
-  meidaportmax:     0
+  mediaip:          "" #媒体服务器地址 默认 自动适配设备网段
+  mediaport:        58200 #媒体服务器端口，用于接收设备的流
+  medianetwork:     tcp
+  mediaportmin:     0 #媒体服务器端口范围最小值，设置后将开启端口范围模式
+  mediaportmax:     0 #媒体服务器端口范围最大值，设置后将开启端口范围模式

-  removebaninterval: 10m
+  removebaninterval: 10m #定时移除注册失败的设备黑名单，单位秒，默认10分钟（600秒）
  loglevel:         info
 ```

- `AutoInvite`     bool 表示自动发起invite，当Server（SIP）接收到设备信息时，立即向设备发送invite命令获取流
- `PreFetchRecord` bool
-
-* sip服务器的配置
- `SipNetwork` string 传输协议，默认UDP，可选TCP
- `SipIP`      string sip 服务器公网IP 默认 自动适配设备网段
- `SipPort`    uint16 sip 服务器端口，默认 5060
- `Serial`     string sip 服务器 id, 默认 34020000002000000001
- `Realm`      string sip 服务器域，默认 3402000000
- `Username`   string sip 服务器账号
- `Password`   string sip 服务器密码
-
- `RegisterValidity`  time.Duration    注册有效期，单位秒，默认 60
-
-* 媒体服务器配置
- `MediaIP`          string 媒体服务器地址 默认 自动适配设备网段
- `MediaPort`        uint16 媒体服务器端口
- `MediaNetwork`     string 媒体传输协议，默认UDP，可选TCP
- `MediaIdleTimeout` uint16 推流超时时间，超过则断开链接，让设备重连
- `MediaPortMin`     uint16 媒体服务器端口范围最小值
- `MediaPortMax`     uint16 媒体服务器端口范围最大值
- `LogLevel`          string 日志级别，默认 info（trace，debug，info，warn，error，fatal, panic）
- `RemoveBanInterval` time.Duration  定时移除注册失败的设备黑名单，单位秒，默认10分钟（600秒）
- `UdpCacheSize`      int  表示UDP缓存大小，默认为0，不开启。仅当TCP关闭，切缓存大于0时才开启，会最多缓存最多N个包，并排序，修复乱序造成的无法播放问题，注意开启后，会有一定的性能损耗，并丢失部分包。
-
 **如果配置了端口范围，将采用范围端口机制，每一个流对应一个端口

 **注意某些摄像机没有设置用户名的地方，摄像机会以自身的国标id作为用户名，这个时候m7s会忽略使用摄像机的用户名，忽略配置的用户名**
--- a/channel.go
+++ b/channel.go
@@ -5,27 +5,55 @@ import (
 	"net/http"
 	"strconv"
 	"strings"
-	"sync"
 	"time"

+	"sync/atomic"
+
 	"github.com/ghettovoice/gosip/sip"
 	"go.uber.org/zap"
 	. "m7s.live/engine/v4"
 	"m7s.live/plugin/gb28181/v4/utils"
+	"m7s.live/plugin/ps/v4"
 )

+type PullStream struct {
+	opt       *InviteOptions
+	channel   *Channel
+	inviteRes sip.Response
+}
+
+func (p *PullStream) Bye() int {
+	res := p.inviteRes
+	bye := p.channel.CreateRequst(sip.BYE)
+	from, _ := res.From()
+	to, _ := res.To()
+	callId, _ := res.CallID()
+	bye.ReplaceHeaders(from.Name(), []sip.Header{from})
+	bye.ReplaceHeaders(to.Name(), []sip.Header{to})
+	bye.ReplaceHeaders(callId.Name(), []sip.Header{callId})
+	resp, err := p.channel.device.SipRequestForResponse(bye)
+	if p.opt.IsLive() {
+		p.channel.status.Store(0)
+		// defer p.channel.TryAutoInvite(p.opt)
+	}
+	if p.opt.recyclePort != nil {
+		p.opt.recyclePort(p.opt.MediaPort)
+	}
+	if err != nil {
+		return ServerInternalError
+	}
+	return int(resp.StatusCode())
+}
+
 type ChannelEx struct {
-	device          *Device
-	RecordPublisher *GBPublisher `json:"-"`
-	LivePublisher   *GBPublisher
-	LiveSubSP       string //实时子码流
+	device          *Device      // 所属设备
+	status          atomic.Int32 // 通道状态,0:空闲,1:正在invite,2:正在播放
+	LiveSubSP       string       // 实时子码流，通过rtsp
 	Records         []*Record
 	RecordStartTime string
 	RecordEndTime   string
 	recordStartTime time.Time
 	recordEndTime   time.Time
-	liveInviteLock  *sync.Mutex
-	tcpPortIndex    uint16
 	GpsTime         time.Time //gps时间
 	Longitude       string    //经度
 	Latitude        string    //纬度
@@ -47,7 +75,7 @@ type Channel struct {
 	RegisterWay  int
 	Secrecy      int
 	Status       string
-	Children     []*Channel `json:"-"`
+	Children     []*Channel `json:"-" yaml:"-"`
 	ChannelEx               //自定义属性
 }

@@ -75,9 +103,13 @@ func (channel *Channel) CreateRequst(Method sip.RequestMethod) (req sip.Request)
 	//非同一域的目标地址需要使用@host
 	host := conf.Realm
 	if channel.DeviceID[0:9] != host {
-		deviceIp := d.NetAddr
-		deviceIp = deviceIp[0:strings.LastIndex(deviceIp, ":")]
-		host = fmt.Sprintf("%s:%d", deviceIp, channel.Port)
+		if channel.Port != 0 {
+			deviceIp := d.NetAddr
+			deviceIp = deviceIp[0:strings.LastIndex(deviceIp, ":")]
+			host = fmt.Sprintf("%s:%d", deviceIp, channel.Port)
+		} else {
+			host = d.NetAddr
+		}
 	}

 	channelAddr := sip.Address{
@@ -199,18 +231,26 @@ f = v/a/编码格式/码率大小/采样率
 f字段中视、音频参数段之间不需空格分割。
 可使用f字段中的分辨率参数标识同一设备不同分辨率的码流。
 */
+
 func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 	if opt.IsLive() {
-		if !channel.liveInviteLock.TryLock() {
+		if !channel.status.CompareAndSwap(0, 1) {
 			return 304, nil
 		}
 		defer func() {
-			if code != OK {
-				channel.liveInviteLock.Unlock()
+			if err != nil {
+				channel.status.Store(0)
+				if conf.InviteMode == 1 {
+					// 5秒后重试
+					time.AfterFunc(time.Second*5, func() {
+						channel.Invite(opt)
+					})
+				}
+			} else {
+				channel.status.Store(2)
 			}
 		}()
 	}
-	channel.Bye(opt.IsLive())
 	d := channel.device
 	streamPath := fmt.Sprintf("%s/%s", d.ID, channel.DeviceID)
 	s := "Play"
@@ -219,37 +259,36 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 		s = "Playback"
 		streamPath = fmt.Sprintf("%s/%s/%d-%d", d.ID, channel.DeviceID, opt.Start, opt.End)
 	}
+	if opt.StreamPath != "" {
+		streamPath = opt.StreamPath
+	}
 	if opt.dump == "" {
 		opt.dump = conf.DumpPath
 	}
-	publisher := &GBPublisher{
-		InviteOptions: opt,
-		channel:       channel,
-	}
-	publisher.DisableReorder = !conf.RtpReorder
 	protocol := ""
+	networkType := "udp"
+	resuePort := true
 	if conf.IsMediaNetworkTCP() {
+		networkType = "tcp"
 		protocol = "TCP/"
 		if conf.tcpPorts.Valid {
-			opt.MediaPort, err = publisher.ListenTCP()
-			if err != nil {
-				return ServerInternalError, err
-			}
-		} else if opt.MediaPort == 0 {
-			opt.MediaPort = conf.MediaPort
+			opt.MediaPort, err = conf.tcpPorts.GetPort()
+			opt.recyclePort = conf.tcpPorts.Recycle
+			resuePort = false
 		}
-		publisher.DisableReorder = true
 	} else {
 		if conf.udpPorts.Valid {
-			opt.MediaPort, err = publisher.ListenUDP()
-			if err != nil {
-				code = ServerInternalError
-				return
-			}
-		} else if opt.MediaPort == 0 {
-			opt.MediaPort = conf.MediaPort
+			opt.MediaPort, err = conf.udpPorts.GetPort()
+			opt.recyclePort = conf.udpPorts.Recycle
+			resuePort = false
 		}
 	}
+	if err != nil {
+		return http.StatusInternalServerError, err
+	}
+	if opt.MediaPort == 0 {
+		opt.MediaPort = conf.MediaPort
+	}

 	sdpInfo := []string{
 		"v=0",
@@ -262,7 +301,6 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 		"a=recvonly",
 		"a=rtpmap:96 PS/90000",
 		"y=" + opt.ssrc,
-		"",
 	}
 	if conf.IsMediaNetworkTCP() {
 		sdpInfo = append(sdpInfo, "a=setup:passive", "a=connection:new")
@@ -271,22 +309,22 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 	contentType := sip.ContentType("application/sdp")
 	invite.AppendHeader(&contentType)

-	invite.SetBody(strings.Join(sdpInfo, "\r\n"), true)
+	invite.SetBody(strings.Join(sdpInfo, "\r\n")+"\r\n", true)

 	subject := sip.GenericHeader{
 		HeaderName: "Subject", Contents: fmt.Sprintf("%s:%s,%s:0", channel.DeviceID, opt.ssrc, conf.Serial),
 	}
 	invite.AppendHeader(&subject)
-	publisher.inviteRes, err = d.SipRequestForResponse(invite)
+	inviteRes, err := d.SipRequestForResponse(invite)
 	if err != nil {
 		plugin.Error(fmt.Sprintf("SIP->Invite %s :%s invite error: %s", channel.DeviceID, invite.String(), err.Error()))
 		return http.StatusInternalServerError, err
 	}
-	code = int(publisher.inviteRes.StatusCode())
+	code = int(inviteRes.StatusCode())
 	plugin.Info(fmt.Sprintf("Channel :%s invite response status code: %d", channel.DeviceID, code))

 	if code == OK {
-		ds := strings.Split(publisher.inviteRes.Body(), "\r\n")
+		ds := strings.Split(inviteRes.Body(), "\r\n")
 		for _, l := range ds {
 			if ls := strings.Split(l, "="); len(ls) > 1 {
 				if ls[0] == "y" && len(ls[1]) > 0 {
@@ -299,46 +337,42 @@ func (channel *Channel) Invite(opt *InviteOptions) (code int, err error) {
 				}
 			}
 		}
-		if conf.UdpCacheSize > 0 && !conf.IsMediaNetworkTCP() {
-			publisher.udpCache = utils.NewPqRtp()
+		err = ps.Receive(streamPath, opt.dump, fmt.Sprintf("%s:%d", networkType, opt.MediaPort), opt.SSRC, resuePort)
+		if err == nil {
+			PullStreams.Store(streamPath, &PullStream{
+				opt:       opt,
+				channel:   channel,
+				inviteRes: inviteRes,
+			})
+			err = srv.Send(sip.NewAckRequest("", invite, inviteRes, "", nil))
 		}
-		if err = plugin.Publish(streamPath, publisher); err != nil {
-			code = ServerInternalError
-			return
-		}
-		ack := sip.NewAckRequest("", invite, publisher.inviteRes, "", nil)
-		srv.Send(ack)
-	} else if channel.CanInvite() {
-		time.AfterFunc(time.Second*5, func() {
-			channel.TryAutoInvite()
-		})
 	}
 	return
 }

-func (channel *Channel) Bye(live bool) int {
+func (channel *Channel) Bye(streamPath string) int {
 	d := channel.device
-	streamPath := fmt.Sprintf("%s/%s", d.ID, channel.DeviceID)
-	if s := Streams.Get(streamPath); s != nil {
-		s.Close()
+	if streamPath == "" {
+		streamPath = fmt.Sprintf("%s/%s", d.ID, channel.DeviceID)
 	}
-	if live && channel.LivePublisher != nil {
-		return channel.LivePublisher.Bye()
+	if s, loaded := PullStreams.LoadAndDelete(streamPath); loaded {
+		s.(*PullStream).Bye()
+		if s := Streams.Get(streamPath); s != nil {
+			s.Close()
+		}
+		return http.StatusOK
 	}
-	if !live && channel.RecordPublisher != nil {
-		return channel.RecordPublisher.Bye()
-	}
-	return 404
+	return http.StatusNotFound
 }

-func (channel *Channel) TryAutoInvite() {
-	if conf.AutoInvite && channel.CanInvite() {
-		go channel.Invite(&InviteOptions{})
+func (channel *Channel) TryAutoInvite(opt *InviteOptions) {
+	if conf.InviteMode == 1 && channel.CanInvite() {
+		go channel.Invite(opt)
 	}
 }

 func (channel *Channel) CanInvite() bool {
-	if channel.LivePublisher != nil || len(channel.DeviceID) != 20 || channel.Status == "OFF" {
+	if channel.status.Load() != 0 || len(channel.DeviceID) != 20 || channel.Status == "OFF" {
 		return false
 	}

--- a/device.go
+++ b/device.go
@@ -10,8 +10,6 @@ import (
 	"sync"
 	"time"

-	"golang.org/x/exp/maps"
-
 	"go.uber.org/zap"
 	"m7s.live/engine/v4"
 	"m7s.live/plugin/gb28181/v4/utils"
@@ -47,7 +45,7 @@ var (
 )

 type Device struct {
-	//*transaction.Core `json:"-"`
+	//*transaction.Core `json:"-" yaml:"-"`
 	ID              string
 	Name            string
 	Manufacturer    string
@@ -62,8 +60,7 @@ type Device struct {
 	sipIP           string //设备对应网卡的服务器ip
 	mediaIP         string //设备对应网卡的服务器ip
 	NetAddr         string
-	ChannelMap      map[string]*Channel
-	channelMutex    sync.RWMutex
+	channelMap      sync.Map
 	subscriber      struct {
 		CallID  string
 		Timeout time.Time
@@ -76,13 +73,18 @@ type Device struct {

 func (d *Device) MarshalJSON() ([]byte, error) {
 	type Alias Device
-	return json.Marshal(&struct {
+	data := &struct {
 		Channels []*Channel
 		*Alias
 	}{
-		Channels: maps.Values(d.ChannelMap),
-		Alias:    (*Alias)(d),
+		Alias: (*Alias)(d),
+	}
+	d.channelMap.Range(func(key, value interface{}) bool {
+		c := value.(*Channel)
+		data.Channels = append(data.Channels, c)
+		return true
 	})
+	return json.Marshal(data)
 }
 func (c *GB28181Config) RecoverDevice(d *Device, req sip.Request) {
 	from, _ := req.From()
@@ -116,7 +118,6 @@ func (c *GB28181Config) RecoverDevice(d *Device, req sip.Request) {
 	d.mediaIP = mediaIp
 	d.NetAddr = deviceIp
 	d.UpdateTime = time.Now()
-	d.ChannelMap = make(map[string]*Channel)
 }

 func (c *GB28181Config) StoreDevice(id string, req sip.Request) (d *Device) {
@@ -162,7 +163,6 @@ func (c *GB28181Config) StoreDevice(id string, req sip.Request) (d *Device) {
 			sipIP:        sipIP,
 			mediaIP:      mediaIp,
 			NetAddr:      deviceIp,
-			ChannelMap:   make(map[string]*Channel),
 		}
 		Devices.Store(id, d)
 		c.SaveDevices()
@@ -198,38 +198,27 @@ func (c *GB28181Config) SaveDevices() {
 }

 func (d *Device) addOrUpdateChannel(channel *Channel) {
-	d.channelMutex.Lock()
-	defer d.channelMutex.Unlock()
+	if old, ok := d.channelMap.Load(channel.DeviceID); ok {
+		channel.ChannelEx = old.(*Channel).ChannelEx
+	}
 	channel.device = d
-	var oldLock *sync.Mutex
-	if old, ok := d.ChannelMap[channel.DeviceID]; ok {
-		//复制锁指针
-		oldLock = old.liveInviteLock
-	}
-	if oldLock == nil {
-		channel.liveInviteLock = &sync.Mutex{}
-	} else {
-		channel.liveInviteLock = oldLock
-	}
-	d.ChannelMap[channel.DeviceID] = channel
+	d.channelMap.Store(channel.DeviceID, channel)
 }

 func (d *Device) deleteChannel(DeviceID string) {
-	d.channelMutex.Lock()
-	defer d.channelMutex.Unlock()
-	delete(d.ChannelMap, DeviceID)
+	d.channelMap.Delete(DeviceID)
 }

 func (d *Device) CheckSubStream() {
-	d.channelMutex.Lock()
-	defer d.channelMutex.Unlock()
-	for _, c := range d.ChannelMap {
+	d.channelMap.Range(func(key, value any) bool {
+		c := value.(*Channel)
 		if s := engine.Streams.Get("sub/" + c.DeviceID); s != nil {
 			c.LiveSubSP = s.Path
 		} else {
 			c.LiveSubSP = ""
 		}
-	}
+		return true
+	})
 }
 func (d *Device) UpdateChannels(list []*Channel) {

@@ -266,7 +255,7 @@ func (d *Device) UpdateChannels(list []*Channel) {
 				go c.QueryRecord(n.Format(TIME_LAYOUT), n.Add(time.Hour*24-time.Second).Format(TIME_LAYOUT))
 			}
 		}
-		c.TryAutoInvite()
+		c.TryAutoInvite(&InviteOptions{})
 		if s := engine.Streams.Get("sub/" + c.DeviceID); s != nil {
 			c.LiveSubSP = s.Path
 		} else {
@@ -275,11 +264,11 @@ func (d *Device) UpdateChannels(list []*Channel) {
 	}
 }
 func (d *Device) UpdateRecord(channelId string, list []*Record) {
-	d.channelMutex.RLock()
-	if c, ok := d.ChannelMap[channelId]; ok {
+	d.channelMap.Range(func(key, value any) bool {
+		c := value.(*Channel)
 		c.Records = append(c.Records, list...)
-	}
-	d.channelMutex.RUnlock()
+		return true
+	})
 }

 func (d *Device) CreateRequest(Method sip.RequestMethod) (req sip.Request) {
@@ -456,7 +445,8 @@ func (d *Device) MobilePositionSubscribe(id string, expires time.Duration, inter

 // UpdateChannelPosition 更新通道GPS坐标
 func (d *Device) UpdateChannelPosition(channelId string, gpsTime string, lng string, lat string) {
-	if c, ok := d.ChannelMap[channelId]; ok {
+	if v, ok := d.channelMap.Load(channelId); ok {
+		c := v.(*Channel)
 		c.ChannelEx.GpsTime = time.Now() //时间取系统收到的时间，避免设备时间和格式问题
 		c.ChannelEx.Longitude = lng
 		c.ChannelEx.Latitude = lat
@@ -535,7 +525,8 @@ func (d *Device) UpdateChannelStatus(deviceList []*notifyMessage) {
 }

 func (d *Device) channelOnline(DeviceID string) {
-	if c, ok := d.ChannelMap[DeviceID]; ok {
+	if v, ok := d.channelMap.Load(DeviceID); ok {
+		c := v.(*Channel)
 		c.Status = "ON"
 		plugin.Sugar().Debugf("通道[%s]在线\n", c.Name)
 	} else {
@@ -544,7 +535,8 @@ func (d *Device) channelOnline(DeviceID string) {
 }

 func (d *Device) channelOffline(DeviceID string) {
-	if c, ok := d.ChannelMap[DeviceID]; ok {
+	if v, ok := d.channelMap.Load(DeviceID); ok {
+		c := v.(*Channel)
 		c.Status = "OFF"
 		plugin.Sugar().Debugf("通道[%s]离线\n", c.Name)
 	} else {
--- a/go.mod
+++ b/go.mod
@@ -9,14 +9,15 @@ require (
 	github.com/pion/rtp v1.7.13
 	go.uber.org/zap v1.23.0
 	golang.org/x/exp v0.0.0-20221205204356-47842c84f3db
-	golang.org/x/net v0.7.0
-	golang.org/x/text v0.7.0
-	m7s.live/engine/v4 v4.11.15
+	golang.org/x/net v0.8.0
+	golang.org/x/text v0.8.0
+	m7s.live/engine/v4 v4.12.0
 )

 require (
-	github.com/aler9/gortsplib/v2 v2.1.4 // indirect
+	github.com/aler9/gortsplib/v2 v2.2.2 // indirect
 	github.com/cnotch/ipchub v1.1.0 // indirect
+	github.com/denisbrodbeck/machineid v1.0.1 // indirect
 	github.com/discoviking/fsm v0.0.0-20150126104936-f4a273feecca // indirect
 	github.com/fsnotify/fsnotify v1.6.0 // indirect
 	github.com/go-ole/go-ole v1.2.6 // indirect
@@ -53,10 +54,10 @@ require (
 	go.uber.org/atomic v1.10.0 // indirect
 	go.uber.org/multierr v1.8.0 // indirect
 	golang.org/x/crypto v0.4.0 // indirect
-	golang.org/x/mod v0.7.0 // indirect
+	golang.org/x/mod v0.8.0 // indirect
 	golang.org/x/sync v0.1.0 // indirect
-	golang.org/x/sys v0.5.0 // indirect
-	golang.org/x/term v0.5.0 // indirect
-	golang.org/x/tools v0.3.0 // indirect
+	golang.org/x/sys v0.6.0 // indirect
+	golang.org/x/term v0.6.0 // indirect
+	golang.org/x/tools v0.6.0 // indirect
 	gopkg.in/yaml.v3 v3.0.1 // indirect
 )
--- a/go.sum
+++ b/go.sum
@@ -1,6 +1,6 @@
 github.com/BurntSushi/toml v0.3.1/go.mod h1:xHWCNGjB5oqiDr8zfno3MHue2Ht5sIBksp03qcyfWMU=
-github.com/aler9/gortsplib/v2 v2.1.4 h1:A4C4Qxz3aQibphXoKsifwKmKZRY7leaO3jHkA+SQ2kw=
-github.com/aler9/gortsplib/v2 v2.1.4/go.mod h1:Eegw8PWa8hNYXiYMlbK3RX1gr7+r25MxniAPGA+kKUE=
+github.com/aler9/gortsplib/v2 v2.2.2 h1:tTw8pdKSOEjlZjjE1S4ftXPHJkYOqjNNv3hjQ0Nto9M=
+github.com/aler9/gortsplib/v2 v2.2.2/go.mod h1:k6uBVHGwsIc/0L5SLLqWwi6bSJUb4VR0HfvncyHlKQI=
 github.com/benbjohnson/clock v1.1.0 h1:Q92kusRqC1XV2MjkWETPvjJVqKetz1OzxZB7mHJLju8=
 github.com/chzyer/logex v1.1.10/go.mod h1:+Ywpsq7O8HXn0nuIou7OrIPyXbp3wmkHB+jjWRnGsAI=
 github.com/chzyer/readline v0.0.0-20180603132655-2972be24d48e/go.mod h1:nSuG5e5PlCu98SY8svDHJxuZscDgtXS6KTTbou5AhLI=
@@ -16,6 +16,8 @@ github.com/cnotch/xlog v0.0.0-20201208005456-cfda439cd3a0/go.mod h1:RW9oHsR79ffl
 github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
 github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
 github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
+github.com/denisbrodbeck/machineid v1.0.1 h1:geKr9qtkB876mXguW2X6TU4ZynleN6ezuMSRhl4D7AQ=
+github.com/denisbrodbeck/machineid v1.0.1/go.mod h1:dJUwb7PTidGDeYyUBmXZ2GphQBbjJCrnectwCyxcUSI=
 github.com/discoviking/fsm v0.0.0-20150126104936-f4a273feecca h1:cTTdXpkQ1aVbOOmHwdwtYuwUZcQtcMrleD1UXLWhAq8=
 github.com/discoviking/fsm v0.0.0-20150126104936-f4a273feecca/go.mod h1:W+3LQaEkN8qAwwcw0KC546sUEnX86GIT8CcMLZC4mG0=
 github.com/emitter-io/address v1.0.0/go.mod h1:GfZb5+S/o8694B1GMGK2imUYQyn2skszMvGNA5D84Ug=
@@ -173,8 +175,8 @@ github.com/stretchr/testify v1.6.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/
 github.com/stretchr/testify v1.7.0/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.7.1/go.mod h1:6Fq8oRcR53rry900zMqJjRRixrwX3KX962/h/Wwjteg=
 github.com/stretchr/testify v1.8.0/go.mod h1:yNjHg4UonilssWZ8iaSj1OCr/vHnekPRkoO+kdMU+MU=
-github.com/stretchr/testify v1.8.1 h1:w7B6lhMri9wdJUVmEZPGGhZzrYTPvgJArz7wNPgYKsk=
 github.com/stretchr/testify v1.8.1/go.mod h1:w2LPCIKwWwSfY2zedu0+kehJoqGctiVI29o6fzry7u4=
+github.com/stretchr/testify v1.8.2 h1:+h33VjcLVPDHtOdpUCuF+7gSuG3yGIftsP1YvFihtJ8=
 github.com/tevino/abool v0.0.0-20170917061928-9b9efcf221b5/go.mod h1:f1SCnEOt6sc3fOJfPQDRDzHOtSXuTtnz0ImG9kPRDV0=
 github.com/tevino/abool v1.2.0 h1:heAkClL8H6w+mK5md9dzsuohKeXHUpY7Vw0ZCKW+huA=
 github.com/tevino/abool v1.2.0/go.mod h1:qc66Pna1RiIsPa7O4Egxxs9OqkuxDX55zznh9K07Tzg=
@@ -215,8 +217,8 @@ golang.org/x/exp v0.0.0-20221205204356-47842c84f3db/go.mod h1:CxIveKay+FTh1D0yPZ
 golang.org/x/mod v0.3.0/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.4.2/go.mod h1:s0Qsj1ACt9ePp/hMypM3fl4fZqREWJwdYDEqhRiZZUA=
 golang.org/x/mod v0.6.0-dev.0.20220419223038-86c51ed26bb4/go.mod h1:jJ57K6gSWd91VN4djpZkiMVwK6gcyfeH4XE8wZrZaV4=
-golang.org/x/mod v0.7.0 h1:LapD9S96VoQRhi/GrNTqeBJFrUjs5UHCAtTlgwA5oZA=
-golang.org/x/mod v0.7.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
+golang.org/x/mod v0.8.0 h1:LUYupSeNrTNCGzR/hVBk2NHZO4hXcVaW1k4Qx7rjPx8=
+golang.org/x/mod v0.8.0/go.mod h1:iBbtSCu2XBx23ZKBPSOrRkjjQPZFPuis4dIYUhu/chs=
 golang.org/x/net v0.0.0-20180906233101-161cd47e91fd/go.mod h1:mL1N/T3taQHkDXs73rZJwtUhF3w3ftmwwsq0BUmARs4=
 golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
 golang.org/x/net v0.0.0-20190620200207-3b0461eec859/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
@@ -234,8 +236,8 @@ golang.org/x/net v0.0.0-20220425223048-2871e0cb64e4/go.mod h1:CfG3xpIq0wQ8r1q4Su
 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.7.0 h1:rJrUqqhjsgNp7KqAIc25s9pZnjU7TUcSY7HcVZjdn1g=
-golang.org/x/net v0.7.0/go.mod h1:2Tu9+aMcznHK/AK1HMvgo6xiTLG5rD5rZLDS+rp2Bjs=
+golang.org/x/net v0.8.0 h1:Zrh2ngAOFYneWTAIAPethzeaQLuHwhuBkuV6ZiRnUaQ=
+golang.org/x/net v0.8.0/go.mod h1:QVkue5JL9kW//ek3r6jTKnTFis1tRmNAW2P1shuFdJc=
 golang.org/x/sync v0.0.0-20180314180146-1d60e4601c6f/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20190423024810-112230192c58/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
 golang.org/x/sync v0.0.0-20201020160332-67f06af15bc9/go.mod h1:RxMgew5VJxzue5/jJTE5uejpjVlOe/izrB70Jof72aM=
@@ -277,29 +279,29 @@ golang.org/x/sys v0.0.0-20220811171246-fbc7d0a398ab/go.mod h1:oPkhp1MJrh7nUepCBc
 golang.org/x/sys v0.0.0-20220908164124-27713097b956/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.5.0 h1:MUK/U/4lj1t1oPg0HfuXDN/Z1wv31ZJ/YcPiGccS4DU=
-golang.org/x/sys v0.5.0/go.mod h1:oPkhp1MJrh7nUepCBck5+mAzfO9JrbApNNgaTdGDITg=
+golang.org/x/sys v0.6.0 h1:MVltZSvRTcU2ljQOhs94SXPftV6DCNnZViHeQps87pQ=
+golang.org/x/sys v0.6.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.5.0 h1:n2a8QNdAb0sZNpU9R1ALUXBbY+w51fCQDN+7EdxNBsY=
-golang.org/x/term v0.5.0/go.mod h1:jMB1sMXY+tzblOD4FWmEbocvup2/aLOaQEp7JmGp78k=
+golang.org/x/term v0.6.0 h1:clScbb1cHjoCkyRbWwBEUZ5H/tIFu5TAXIqaZD0Gcjw=
+golang.org/x/term v0.6.0/go.mod h1:m6U89DPEgQRMq3DNkDClhWw02AUbt2daBVO4cn4Hv9U=
 golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
 golang.org/x/text v0.3.2/go.mod h1:bEr9sfX3Q8Zfm5fL9x+3itogRgK3+ptLWKqgva+5dAk=
 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.7.0 h1:4BRB4x83lYWy72KwLD/qYDuTu7q9PjSagHvijDw7cLo=
-golang.org/x/text v0.7.0/go.mod h1:mrYo+phRRbMaCq/xk9113O4dZlRixOauAjOtrjsXDZ8=
+golang.org/x/text v0.8.0 h1:57P1ETyNKtuIjB4SRd15iJxuhj8Gc416Y78H3qgMh68=
+golang.org/x/text v0.8.0/go.mod h1:e1OnstbJyHTd6l/uOt8jFFHp6TRDWZR/bV3emEE/zU8=
 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=
 golang.org/x/tools v0.1.1/go.mod h1:o0xws9oXOQQZyjljx8fwUC0k7L1pTE6eaCbjGeHmOkk=
 golang.org/x/tools v0.1.12/go.mod h1:hNGJHUnrk76NpqgfD5Aqm5Crs+Hm0VOH/i9J2+nxYbc=
-golang.org/x/tools v0.3.0 h1:SrNbZl6ECOS1qFzgTdQfWXZM9XBkiA6tkFrH9YSTPHM=
-golang.org/x/tools v0.3.0/go.mod h1:/rWhSS2+zyEVwoJf8YAX6L2f0ntZ7Kn/mGgAWcipA5k=
+golang.org/x/tools v0.6.0 h1:BOw41kyTf3PuCW1pVQf8+Cyg8pMlkYB1oo9iJ6D/lKM=
+golang.org/x/tools v0.6.0/go.mod h1:Xwgl3UAJ/d3gWutnCtw505GrjyAbvKui8lOU390QaIU=
 golang.org/x/xerrors v0.0.0-20190717185122-a985d3407aa7/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191011141410-1b5146add898/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
 golang.org/x/xerrors v0.0.0-20191204190536-9bdfabe68543/go.mod h1:I/5z698sn9Ka8TeJc9MKroUUfqBBauWjQqLJ2OPfmY0=
@@ -330,5 +332,5 @@ gopkg.in/yaml.v3 v3.0.0-20200313102051-9f266ea9e77c/go.mod h1:K4uyk7z7BCEPqu6E+C
 gopkg.in/yaml.v3 v3.0.0-20210107192922-496545a6307b/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
 gopkg.in/yaml.v3 v3.0.1 h1:fxVm/GzAzEWqLHuvctI91KS9hhNmmWOoWu0XTYJS7CA=
 gopkg.in/yaml.v3 v3.0.1/go.mod h1:K4uyk7z7BCEPqu6E+C64Yfv1cQ7kz7rIZviUmN+EgEM=
-m7s.live/engine/v4 v4.11.15 h1:Hwcfsw1XK63tSJlt7oI+2NY+DXds1urSZJtjQqeB0T8=
-m7s.live/engine/v4 v4.11.15/go.mod h1:0gK75fj3GjUcVX5Tu/zC7MSob5nFnA1BYTeMt3w7uMU=
+m7s.live/engine/v4 v4.12.0 h1:CRPbJ0jhHVZArc5mvV7e6Seb4Ye816kGzs3FOVKnfHw=
+m7s.live/engine/v4 v4.12.0/go.mod h1:AiJPBwdA77DM3fymlcH2qYPR8ivL6ib9UVLm1Rft/to=
--- a/handle.go
+++ b/handle.go
@@ -185,12 +185,14 @@ func (c *GB28181Config) OnMessage(req sip.Request, tx sip.ServerTransaction) {
 		case "Keepalive":
 			d.LastKeepaliveAt = time.Now()
 			//callID !="" 说明是订阅的事件类型信息
-			if d.ChannelMap == nil || len(d.ChannelMap) == 0 {
+			if d.lastSyncTime.IsZero() {
 				go d.syncChannels()
 			} else {
-				for _, ch := range d.ChannelMap {
-					ch.TryAutoInvite()
-				}
+				d.channelMap.Range(func(key, value interface{}) bool {
+					channel := value.(*Channel)
+					channel.TryAutoInvite(&InviteOptions{})
+					return true
+				})
 			}
 			//为什么要查找子码流?
 			//d.CheckSubStream()
--- a/inviteoption.go
+++ b/inviteoption.go
@@ -8,12 +8,14 @@ import (
 )

 type InviteOptions struct {
-	Start     int
-	End       int
-	dump      string
-	ssrc      string
-	SSRC      uint32
-	MediaPort uint16
+	Start       int
+	End         int
+	dump        string
+	ssrc        string
+	SSRC        uint32
+	MediaPort   uint16
+	StreamPath  string
+	recyclePort func(p uint16) (err error)
 }

 func (o InviteOptions) IsLive() bool {
--- a/main.go
+++ b/main.go
@@ -2,12 +2,14 @@ package gb28181

 import (
 	"fmt"
+	"os"
 	"strings"
+	"sync"
 	"time"

 	myip "github.com/husanpao/ip"
 	. "m7s.live/engine/v4"
-	"m7s.live/engine/v4/config"
+	"m7s.live/engine/v4/util"
 )

 type GB28181PositionConfig struct {
@@ -17,19 +19,23 @@ type GB28181PositionConfig struct {
 }

 type GB28181Config struct {
-	AutoInvite     bool `default:"true"`
+	// AutoInvite     bool `default:"true"`
+	InviteMode     int `default:"1"` //邀请模式，0:手动拉流，1:预拉流，2:按需拉流
 	PreFetchRecord bool
 	InviteIDs      string //按照国标gb28181协议允许邀请的设备类型:132 摄像机 NVR
-
+	ListenAddr     string `default:"0.0.0.0"`
 	//sip服务器的配置
-	SipNetwork string `default:"udp"` //传输协议，默认UDP，可选TCP
-	SipIP      string //sip 服务器公网IP
-	SipPort    uint16 `default:"5060"`                 //sip 服务器端口，默认 5060
-	Serial     string `default:"34020000002000000001"` //sip 服务器 id, 默认 34020000002000000001
-	Realm      string `default:"3402000000"`           //sip 服务器域，默认 3402000000
-	Username   string //sip 服务器账号
-	Password   string //sip 服务器密码
-
+	SipNetwork string   `default:"udp"` //传输协议，默认UDP，可选TCP
+	SipIP      string   //sip 服务器公网IP
+	SipPort    uint16   `default:"5060"`                 //sip 服务器端口，默认 5060
+	Serial     string   `default:"34020000002000000001"` //sip 服务器 id, 默认 34020000002000000001
+	Realm      string   `default:"3402000000"`           //sip 服务器域，默认 3402000000
+	Username   string   //sip 服务器账号
+	Password   string   //sip 服务器密码
+	Port       struct { // 新配置方式
+		Sip   string `default:"udp:5060"`
+		Media string `default:"tcp:58200"`
+	}
 	// AckTimeout        uint16 //sip 服务应答超时，单位秒
 	RegisterValidity time.Duration `default:"60s"` //注册有效期，单位秒，默认 3600
 	// RegisterInterval  int    //注册间隔，单位秒，默认 60
@@ -39,22 +45,23 @@ type GB28181Config struct {
 	//媒体服务器配置
 	MediaIP      string //媒体服务器地址
 	MediaPort    uint16 `default:"58200"` //媒体服务器端口
-	MediaNetwork string `default:"udp"`   //媒体传输协议，默认UDP，可选TCP
+	MediaNetwork string `default:"tcp"`   //媒体传输协议，默认UDP，可选TCP
 	MediaPortMin uint16
 	MediaPortMax uint16
 	// MediaIdleTimeout uint16 //推流超时时间，超过则断开链接，让设备重连

 	// WaitKeyFrame      bool //是否等待关键帧，如果等待，则在收到第一个关键帧之前，忽略所有媒体流
 	RemoveBanInterval time.Duration `default:"600s"` //移除禁止设备间隔
-	UdpCacheSize      int           //udp缓存大小
-	LogLevel          string        `default:"info"` //trace, debug, info, warn, error, fatal, panic
-	routes            map[string]string
-	DumpPath          string //dump PS流本地文件路径
-	RtpReorder        bool   `default:"true"`
-	config.Publish
-	Server
+	// UdpCacheSize      int           //udp缓存大小
+	LogLevel string `default:"info"` //trace, debug, info, warn, error, fatal, panic
+	routes   map[string]string
+	DumpPath string //dump PS流本地文件路径
+	Ignores  map[string]struct{}
+	tcpPorts PortManager
+	udpPorts PortManager

 	Position GB28181PositionConfig //关于定位的配置参数
+
 }

 func (c *GB28181Config) initRoutes() {
@@ -68,12 +75,39 @@ func (c *GB28181Config) initRoutes() {
 	}
 	plugin.Info(fmt.Sprintf("LocalAndInternalIPs detail: %s", c.routes))
 }
+
 func (c *GB28181Config) OnEvent(event any) {
-	switch event.(type) {
+	switch e := event.(type) {
 	case FirstConfig:
+		if c.Port.Sip != "udp:5060" {
+			protocol, ports := util.Conf2Listener(c.Port.Sip)
+			c.SipNetwork = protocol
+			c.SipPort = ports[0]
+		}
+		if c.Port.Media != "tcp:58200" {
+			protocol, ports := util.Conf2Listener(c.Port.Media)
+			c.MediaNetwork = protocol
+			if len(ports) > 1 {
+				c.MediaPortMin = ports[0]
+				c.MediaPortMax = ports[1]
+			} else {
+				c.MediaPort = ports[0]
+			}
+		}
+		os.MkdirAll(c.DumpPath, 0766)
 		c.ReadDevices()
 		go c.initRoutes()
 		c.startServer()
+	case *Stream:
+		if c.InviteMode == 2 {
+			if channel := FindChannel(e.AppName, e.StreamName); channel != nil {
+				channel.TryAutoInvite(&InviteOptions{})
+			}
+		}
+	case SEclose:
+		if v, ok := PullStreams.LoadAndDelete(e.Target.Path); ok {
+			go v.(*PullStream).Bye()
+		}
 	}
 }

@@ -84,3 +118,4 @@ func (c *GB28181Config) IsMediaNetworkTCP() bool {
 var conf GB28181Config

 var plugin = InstallPlugin(&conf)
+var PullStreams sync.Map //拉流
--- a/publisher.go
+++ b/publisher.go
@@ -1,229 +0,0 @@
-package gb28181
-
-import (
-	"encoding/binary"
-	"fmt"
-	"io"
-	"net"
-	"os"
-	"time"
-
-	"github.com/ghettovoice/gosip/sip"
-	"github.com/pion/rtp"
-	"go.uber.org/zap"
-	. "m7s.live/engine/v4"
-	"m7s.live/engine/v4/util"
-	"m7s.live/plugin/gb28181/v4/utils"
-)
-
-type GBPublisher struct {
-	PSPublisher
-	*InviteOptions
-	channel     *Channel
-	inviteRes   sip.Response
-	udpCache    *utils.PriorityQueueRtp
-	dumpFile    *os.File
-	dumpPrint   io.Writer
-	lastReceive time.Time
-}
-
-func (p *GBPublisher) PrintDump(s string) {
-	if p.dumpPrint != nil {
-		p.dumpPrint.Write([]byte(s))
-	}
-}
-
-func (p *GBPublisher) OnEvent(event any) {
-	if p.channel == nil {
-		// p.parser.EsHandler = p
-		p.IO.OnEvent(event)
-		return
-	}
-	switch event.(type) {
-	case IPublisher:
-		if p.IsLive() {
-			p.Type = "GB28181 Live"
-			p.channel.LivePublisher = p
-		} else {
-			p.Type = "GB28181 Playback"
-			p.channel.RecordPublisher = p
-		}
-		// p.parser.EsHandler = p
-		conf.publishers.Add(p.SSRC, p)
-		if err := error(nil); p.dump != "" {
-			if p.dumpFile, err = os.OpenFile(p.dump, os.O_WRONLY|os.O_CREATE, 0644); err != nil {
-				p.Error("open dump file failed", zap.Error(err))
-			}
-		}
-	case SEwaitPublish:
-		//掉线自动重新拉流
-		if p.IsLive() {
-			if p.channel.LivePublisher != nil {
-				p.channel.LivePublisher = nil
-				p.channel.liveInviteLock.Unlock()
-			}
-			go p.channel.Invite(&InviteOptions{})
-		}
-	case SEclose, SEKick:
-		if p.IsLive() {
-			if p.channel.LivePublisher != nil {
-				p.channel.LivePublisher = nil
-				p.channel.liveInviteLock.Unlock()
-			}
-		} else {
-			p.channel.RecordPublisher = nil
-		}
-		conf.publishers.Delete(p.SSRC)
-		if p.dumpFile != nil {
-			p.dumpFile.Close()
-		}
-		p.Bye()
-	}
-	p.Publisher.OnEvent(event)
-}
-
-func (p *GBPublisher) Bye() int {
-	res := p.inviteRes
-	if res == nil {
-		return 404
-	}
-	defer p.Stop()
-	p.inviteRes = nil
-	bye := p.channel.CreateRequst(sip.BYE)
-	from, _ := res.From()
-	to, _ := res.To()
-	callId, _ := res.CallID()
-	bye.ReplaceHeaders(from.Name(), []sip.Header{from})
-	bye.ReplaceHeaders(to.Name(), []sip.Header{to})
-	bye.ReplaceHeaders(callId.Name(), []sip.Header{callId})
-	resp, err := p.channel.device.SipRequestForResponse(bye)
-	if err != nil {
-		p.Error("Bye", zap.Error(err))
-		return ServerInternalError
-	}
-	return int(resp.StatusCode())
-}
-
-func (p *GBPublisher) Replay(f *os.File) (err error) {
-	var rtpPacket rtp.Packet
-	defer f.Close()
-	if p.dumpPrint != nil {
-		p.PrintDump(`<style  type="text/css">
-		.gray {
-			color: gray;
-		}
-		</style>
-		`)
-		p.PrintDump("<table>")
-		defer p.PrintDump("</table>")
-	}
-	var t uint16
-	for l := make([]byte, 6); !p.IsClosed(); time.Sleep(time.Millisecond * time.Duration(t)) {
-		_, err = f.Read(l)
-		if err != nil {
-			return
-		}
-		payload := make([]byte, util.ReadBE[int](l[:4]))
-		t = util.ReadBE[uint16](l[4:])
-		p.PrintDump(fmt.Sprintf("[<b>%d</b> %d]", t, len(payload)))
-		_, err = f.Read(payload)
-		if err != nil {
-			return
-		}
-		rtpPacket.Unmarshal(payload)
-		p.PushPS(&rtpPacket)
-	}
-	return
-}
-
-func (p *GBPublisher) ListenUDP() (port uint16, err error) {
-	var rtpPacket rtp.Packet
-	networkBuffer := 1048576
-	port, err = conf.udpPorts.GetPort()
-	if err != nil {
-		return
-	}
-	addr := fmt.Sprintf(":%d", port)
-	mediaAddr, _ := net.ResolveUDPAddr("udp", addr)
-	conn, err := net.ListenUDP("udp", mediaAddr)
-	if err != nil {
-		conf.udpPorts.Recycle(port)
-		plugin.Error("listen media server udp err", zap.String("addr", addr), zap.Error(err))
-		return 0, err
-	}
-	p.SetIO(conn)
-	go func() {
-		defer conn.Close()
-		bufUDP := make([]byte, networkBuffer)
-		plugin.Info("Media udp server start.", zap.Uint16("port", port))
-		defer plugin.Info("Media udp server stop", zap.Uint16("port", port))
-		defer conf.udpPorts.Recycle(port)
-		dumpLen := make([]byte, 6)
-		conn.SetReadDeadline(time.Now().Add(time.Second * 10))
-		for n, _, err := conn.ReadFromUDP(bufUDP); err == nil; n, _, err = conn.ReadFromUDP(bufUDP) {
-			ps := bufUDP[:n]
-			if err := rtpPacket.Unmarshal(ps); err != nil {
-				plugin.Error("Decode rtp error:", zap.Error(err))
-			}
-			if p.dumpFile != nil {
-				util.PutBE(dumpLen[:4], n)
-				if p.lastReceive.IsZero() {
-					util.PutBE(dumpLen[4:], 0)
-				} else {
-					util.PutBE(dumpLen[4:], uint16(time.Since(p.lastReceive).Milliseconds()))
-				}
-				p.lastReceive = time.Now()
-				p.dumpFile.Write(dumpLen)
-				p.dumpFile.Write(ps)
-			}
-			p.PushPS(&rtpPacket)
-			conn.SetReadDeadline(time.Now().Add(time.Second * 10))
-		}
-	}()
-	return
-}
-
-func (p *GBPublisher) ListenTCP() (port uint16, err error) {
-	port, err = conf.tcpPorts.GetPort()
-	if err != nil {
-		return
-	}
-	addr := fmt.Sprintf(":%d", port)
-	mediaAddr, _ := net.ResolveTCPAddr("tcp", addr)
-	listen, err := net.ListenTCP("tcp", mediaAddr)
-	if err != nil {
-		defer conf.tcpPorts.Recycle(port)
-		plugin.Error("listen media server tcp err", zap.String("addr", addr), zap.Error(err))
-		return 0, err
-	}
-	go func() {
-		plugin.Info("Media tcp server start.", zap.Uint16("port", port))
-		defer conf.tcpPorts.Recycle(port)
-		defer plugin.Info("Media tcp server stop", zap.Uint16("port", port))
-		conn, err := listen.Accept()
-		listen.Close()
-		p.SetIO(conn)
-		if err != nil {
-			plugin.Error("Accept err=", zap.Error(err))
-			return
-		}
-		var rtpPacket rtp.Packet
-		lenBuf := make([]byte, 2)
-		defer conn.Close()
-		for err == nil {
-			if _, err = io.ReadFull(conn, lenBuf); err != nil {
-				return
-			}
-			ps := make([]byte, binary.BigEndian.Uint16(lenBuf))
-			if _, err = io.ReadFull(conn, ps); err != nil {
-				return
-			}
-			if err := rtpPacket.Unmarshal(ps); err != nil {
-				plugin.Error("gb28181 decode rtp error:", zap.Error(err))
-			} else if !p.IsClosed() {
-				p.PushPS(&rtpPacket)
-			}
-		}
-	}()
-	return
-}
--- a/restful.go
+++ b/restful.go
@@ -2,9 +2,7 @@ package gb28181

 import (
 	"net/http"
-	"os"
 	"strconv"
-	"strings"
 	"time"

 	"m7s.live/engine/v4/util"
@@ -52,15 +50,17 @@ func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) {
 	query := r.URL.Query()
 	id := query.Get("id")
 	channel := query.Get("channel")
+	streamPath := query.Get("streamPath")
 	port, _ := strconv.Atoi(query.Get("mediaPort"))
 	opt := InviteOptions{
-		dump:      query.Get("dump"),
-		MediaPort: uint16(port),
+		dump:       query.Get("dump"),
+		MediaPort:  uint16(port),
+		StreamPath: streamPath,
 	}
 	opt.Validate(query.Get("startTime"), query.Get("endTime"))
 	if c := FindChannel(id, channel); c == nil {
 		http.NotFound(w, r)
-	} else if opt.IsLive() && c.LivePublisher != nil {
+	} else if opt.IsLive() && c.status.Load() > 0 {
 		w.WriteHeader(304) //直播流已存在
 	} else if code, err := c.Invite(&opt); err == nil {
 		w.WriteHeader(code)
@@ -69,46 +69,12 @@ func (c *GB28181Config) API_invite(w http.ResponseWriter, r *http.Request) {
 	}
 }

-func (c *GB28181Config) API_replay(w http.ResponseWriter, r *http.Request) {
-	dump := r.URL.Query().Get("dump")
-	printOut := r.URL.Query().Get("print")
-	if dump == "" {
-		dump = c.DumpPath
-	}
-	f, err := os.OpenFile(dump, os.O_RDONLY, 0644)
-	if err != nil {
-		http.Error(w, err.Error(), http.StatusInternalServerError)
-	} else {
-		streamPath := dump
-		if strings.HasPrefix(dump, "/") {
-			streamPath = "replay" + dump
-		} else {
-			streamPath = "replay/" + dump
-		}
-		var pub GBPublisher
-		pub.SetIO(f)
-		if err = plugin.Publish(streamPath, &pub); err == nil {
-			if printOut != "" {
-				pub.dumpPrint = w
-				pub.SetParentCtx(r.Context())
-				err = pub.Replay(f)
-			} else {
-				go pub.Replay(f)
-				w.Write([]byte("ok"))
-			}
-		} else {
-			http.Error(w, err.Error(), http.StatusInternalServerError)
-		}
-	}
-}
-
 func (c *GB28181Config) API_bye(w http.ResponseWriter, r *http.Request) {
-	// CORS(w, r)
 	id := r.URL.Query().Get("id")
 	channel := r.URL.Query().Get("channel")
-	live := r.URL.Query().Get("live")
+	streamPath := r.URL.Query().Get("streamPath")
 	if c := FindChannel(id, channel); c != nil {
-		w.WriteHeader(c.Bye(live != "false"))
+		w.WriteHeader(c.Bye(streamPath))
 	} else {
 		http.NotFound(w, r)
 	}
--- a/server.go
+++ b/server.go
@@ -1,20 +1,14 @@
 package gb28181

 import (
-	"bufio"
 	"context"
-	"encoding/binary"
 	"fmt"
-	"io"
-	"net"
 	"strconv"
 	"strings"
 	"time"

 	"github.com/logrusorgru/aurora"
-	"github.com/pion/rtp"
 	"go.uber.org/zap"
-	"m7s.live/engine/v4/util"
 	"m7s.live/plugin/gb28181/v4/utils"

 	"github.com/ghettovoice/gosip"
@@ -24,21 +18,14 @@ import (

 var srv gosip.Server

-type Server struct {
-	Ignores    map[string]struct{}
-	publishers util.Map[uint32, *GBPublisher]
-	tcpPorts   PortManager
-	udpPorts   PortManager
-}
-
 const MaxRegisterCount = 3

 func FindChannel(deviceId string, channelId string) (c *Channel) {
 	if v, ok := Devices.Load(deviceId); ok {
 		d := v.(*Device)
-		d.channelMutex.RLock()
-		c = d.ChannelMap[channelId]
-		d.channelMutex.RUnlock()
+		if v, ok := d.channelMap.Load(channelId); ok {
+			return v.(*Channel)
+		}
 	}
 	return
 }
@@ -129,8 +116,7 @@ func RequestForResponse(transport string, request sip.Request,
 }

 func (c *GB28181Config) startServer() {
-	c.publishers.Init()
-	addr := "0.0.0.0:" + strconv.Itoa(int(c.SipPort))
+	addr := c.ListenAddr + ":" + strconv.Itoa(int(c.SipPort))

 	logger := utils.NewZapLogger(plugin.Logger, "GB SIP Server", nil)
 	logger.SetLevel(levelMap[c.LogLevel])
@@ -151,111 +137,17 @@ func (c *GB28181Config) startServer() {
 		plugin.Info(fmt.Sprint(aurora.Green("Server gb28181 start at"), aurora.BrightBlue(addr)))
 	}

-	go c.startMediaServer()
+	if c.MediaNetwork == "tcp" {
+		c.tcpPorts.Init(c.MediaPortMin, c.MediaPortMax)
+	} else {
+		c.udpPorts.Init(c.MediaPortMin, c.MediaPortMax)
+	}

 	if c.Username != "" || c.Password != "" {
 		go c.removeBanDevice()
 	}
 }

-func (c *GB28181Config) startMediaServer() {
-	if c.MediaNetwork == "tcp" {
-		c.tcpPorts.Init(c.MediaPortMin, c.MediaPortMax)
-		if !c.tcpPorts.Valid {
-			c.listenMediaTCP()
-		}
-	} else {
-		c.udpPorts.Init(c.MediaPortMin, c.MediaPortMax)
-		if !c.udpPorts.Valid {
-			c.listenMediaUDP()
-		}
-	}
-}
-
-func (c *GB28181Config) processTcpMediaConn(conn net.Conn) {
-	var rtpPacket rtp.Packet
-	reader := bufio.NewReader(conn)
-	lenBuf := make([]byte, 2)
-	defer conn.Close()
-	var err error
-	ps := make(util.Buffer, 1024)
-	for err == nil {
-		if _, err = io.ReadFull(reader, lenBuf); err != nil {
-			return
-		}
-		ps.Reset()
-		ps.Glow(int(binary.BigEndian.Uint16(lenBuf)))
-		if _, err = io.ReadFull(reader, ps); err != nil {
-			return
-		}
-		if err := rtpPacket.Unmarshal(ps); err != nil {
-			plugin.Error("gb28181 decode rtp error:", zap.Error(err))
-		} else if publisher := c.publishers.Get(rtpPacket.SSRC); publisher != nil && publisher.Publisher.Err() == nil {
-			publisher.PushPS(&rtpPacket)
-		}
-	}
-}
-
-func (c *GB28181Config) listenMediaTCP() {
-	addr := ":" + strconv.Itoa(int(c.MediaPort))
-	mediaAddr, _ := net.ResolveTCPAddr("tcp", addr)
-	listen, err := net.ListenTCP("tcp", mediaAddr)
-
-	if err != nil {
-		plugin.Error("MediaServer listened　tcp err", zap.String("addr", addr), zap.Error(err))
-		return
-	}
-	plugin.Sugar().Infof("MediaServer started tcp at %s", addr)
-	defer listen.Close()
-	defer plugin.Info("MediaServer stopped tcp at", zap.Uint16("port", c.MediaPort))
-
-	for {
-		conn, err := listen.Accept()
-		if err != nil {
-			plugin.Error("Accept err=", zap.Error(err))
-		}
-		go c.processTcpMediaConn(conn)
-	}
-}
-
-func (c *GB28181Config) listenMediaUDP() {
-	var rtpPacket rtp.Packet
-	networkBuffer := 1048576
-
-	addr := ":" + strconv.Itoa(int(c.MediaPort))
-	mediaAddr, _ := net.ResolveUDPAddr("udp", addr)
-	conn, err := net.ListenUDP("udp", mediaAddr)
-
-	if err != nil {
-		plugin.Error(" MediaServer started listening udp err", zap.String("addr", addr), zap.Error(err))
-		return
-	}
-	bufUDP := make([]byte, networkBuffer)
-	plugin.Sugar().Infof("MediaServer started at udp %s", addr)
-	defer plugin.Sugar().Infof("MediaServer stopped at udp %s", addr)
-	dumpLen := make([]byte, 6)
-	for n, _, err := conn.ReadFromUDP(bufUDP); err == nil; n, _, err = conn.ReadFromUDP(bufUDP) {
-		ps := bufUDP[:n]
-		if err := rtpPacket.Unmarshal(ps); err != nil {
-			plugin.Error("Decode rtp error:", zap.Error(err))
-		}
-		if publisher := c.publishers.Get(rtpPacket.SSRC); publisher != nil && publisher.Publisher.Err() == nil {
-			if publisher.dumpFile != nil {
-				util.PutBE(dumpLen[:4], n)
-				if publisher.lastReceive.IsZero() {
-					util.PutBE(dumpLen[4:], 0)
-				} else {
-					util.PutBE(dumpLen[4:], uint16(time.Since(publisher.lastReceive).Milliseconds()))
-				}
-				publisher.lastReceive = time.Now()
-				publisher.dumpFile.Write(dumpLen)
-				publisher.dumpFile.Write(ps)
-			}
-			publisher.PushPS(&rtpPacket)
-		}
-	}
-}
-
 // func queryCatalog(config *transaction.Config) {
 // 	t := time.NewTicker(time.Duration(config.CatalogInterval) * time.Second)
 // 	for range t.C {
--- a/utils/log.go
+++ b/utils/log.go
@@ -6,17 +6,18 @@ import (
 	"github.com/ghettovoice/gosip/log"
 	"go.uber.org/zap"
 	"go.uber.org/zap/zapcore"
+	m7slog "m7s.live/engine/v4/log"
 )

 type ZapLogger struct {
-	log     *zap.Logger
+	log     *m7slog.Logger
 	prefix  string
 	fields  log.Fields
 	sugared *zap.SugaredLogger
 	level   log.Level
 }

-func NewZapLogger(log *zap.Logger, prefix string, fields log.Fields) (z *ZapLogger) {
+func NewZapLogger(log *m7slog.Logger, prefix string, fields log.Fields) (z *ZapLogger) {
 	z = &ZapLogger{
 		log:    log,
 		prefix: prefix,
Author	SHA1	Message	Date
langhuihui	37fd121d11	feat: change to use ps plugin	2023-05-14 11:12:25 +08:00
langhuihui	05fd8c38f7	feat: add a new way to config port	2023-05-04 09:57:20 +08:00
dexter	2d85e46a8b	ChannelEx	2023-04-09 11:08:23 +08:00
charlestamz	4a90d7bf91	修复Recover导致断流的问题	2023-04-05 22:25:53 +08:00
dexter	a020f3ea81	适配引擎修改	2023-04-04 20:20:27 +08:00
dexter	c68862160f	sip监听网卡地址可配，dump 可同时多路，修复 tcp 方式，媒体默认使用 tcp	2023-03-28 19:42:37 +08:00