万路并发

pkg/gbs/catalog.go

@@ -71,7 +71,7 @@ func (g GB28181API) sipMessageCatalog(ctx *sip.Context) {
 func (g *GB28181API) QueryCatalog(deviceID string) error {
 	slog.Debug("QueryCatalog", "deviceID", deviceID)
 	ipc, ok := g.svr.memoryStorer.Load(deviceID)
-	if !ok {
+	if !ok || !ipc.IsOnline {
 		return ErrDeviceOffline
 	}
 
@@ -100,10 +100,10 @@ func (s *Server) wrapRequest(t Targeter, method string, contentType *sip.Content
 
 	hb := sip.NewHeaderBuilder().
 		SetTo(to).
-		SetFrom(s.fromAddress).
+		SetFrom(&s.fromAddress).
 		SetContentType(contentType).
 		SetMethod(method).
-		SetContact(s.fromAddress).
+		SetContact(&s.fromAddress).
 		AddVia(&sip.ViaHop{
 			Params: sip.NewParams().Add("branch", sip.String{Str: sip.GenerateBranch()}),
 		})

pkg/gbs/config.go

@@ -96,7 +96,7 @@ func NewBasicParamRequest(sn int32, deviceID string) []byte {
 func (g *GB28181API) QueryConfigDownloadBasic(deviceID string) error {
 	slog.Debug("QueryConfigDownloadBasic", "deviceID", deviceID)
 	ipc, ok := g.svr.memoryStorer.Load(deviceID)
-	if !ok {
+	if !ok || !ipc.IsOnline {
 		return ErrDeviceOffline
 	}
 

pkg/gbs/register.go

@@ -183,9 +183,7 @@ func (g GB28181API) login(ctx *sip.Context, expire string) {
 func (g GB28181API) logout(deviceID string, changeFn func(*gb28181.Device)) error {
 	slog.Info("status change 设备离线", "device_id", deviceID)
 	return g.svr.memoryStorer.Change(deviceID, changeFn, func(d *Device) {
-		d.conn = nil
-		d.source = nil
-		d.to = nil
 		d.Expires = 0
+		d.IsOnline = false
 	})
 }

pkg/gbs/server.go

@@ -38,7 +38,7 @@ type Server struct {
 	gb           *GB28181API
 	mediaService sms.Core
 
-	fromAddress  *sip.Address
+	fromAddress  sip.Address
 	memoryStorer MemoryStorer
 }
 
@@ -67,7 +67,7 @@ func NewServer(cfg *conf.Bootstrap, store gb28181.GB28181, sc sms.Core) (*Server
 	c := Server{
 		Server:       svr,
 		mediaService: sc,
-		fromAddress:  &from,
+		fromAddress:  from,
 		gb:           api,
 		memoryStorer: store.Store().(MemoryStorer),
 	}

pkg/gbs/sip/collector.go

@@ -47,8 +47,8 @@ type NoRepeatFn[T any] func(*T, *T) bool
 func NewCollector[T any](noRepeatFn NoRepeatFn[T]) *Collector[T] {
 	return &Collector[T]{
 		data:       make(map[string]*Content[T]),
-		msg:        make(chan *CollectorMsg[T], 10),
-		createCh:   make(chan string, 10),
+		msg:        make(chan *CollectorMsg[T], 512),
+		createCh:   make(chan string, 100),
 		noRepeatFn: noRepeatFn,
 		observer:   NewObserver(),
 	}