gogrlx/nats-server
1
0
Fork 0
mirror of https://github.com/gogrlx/nats-server.git synced 2026-04-02 03:38:42 -07:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #2065 from alexpantyukhin/extract_server_update

Browse Source 
extract_server_update
This commit is contained in:
Ivan Kozlovic
2021-04-08 09:01:51 -06:00
committed by GitHub
parent 6024c85c09 e16bebb8df
commit c7f8296a85
2 changed files with 12 additions and 15 deletions
Download Patch File Download Diff File Expand all files Collapse all files

18
server/client.go
View File

@@ -2522,11 +2522,7 @@ func (c *client) addShadowSub(sub *subscription, ime *ime) (*subscription, error
}
// Update our route map here.
c.srv.updateRouteSubscriptionMap(im.acc, &nsub, 1)
if c.srv.gateway.enabled {
c.srv.gatewayUpdateSubInterest(im.acc.Name, &nsub, 1)
}
c.srv.updateLeafNodes(im.acc, &nsub, 1)
c.srv.updateRemoteSubscription(im.acc, &nsub, 1)
return &nsub, nil
}
@@ -2972,11 +2968,7 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
// Due to defer, reverse the code order so that execution
// is consistent with other cases where we unsubscribe.
if shouldForward {
if srv.gateway.enabled {
defer srv.gatewayUpdateSubInterest(client.acc.Name, sub, -1)
}
defer srv.updateRouteSubscriptionMap(client.acc, sub, -1)
defer srv.updateLeafNodes(client.acc, sub, -1)
defer srv.updateRemoteSubscription(client.acc, sub, -1)
}
defer client.unsubscribe(client.acc, sub, true, true)
} else if sub.nm > sub.max {
@@ -2984,11 +2976,7 @@ func (c *client) deliverMsg(sub *subscription, subject, reply, mh, msg []byte, g
client.mu.Unlock()
client.unsubscribe(client.acc, sub, true, true)
if shouldForward {
srv.updateRouteSubscriptionMap(client.acc, sub, -1)
if srv.gateway.enabled {
srv.gatewayUpdateSubInterest(client.acc.Name, sub, -1)
}
srv.updateLeafNodes(client.acc, sub, -1)
srv.updateRemoteSubscription(client.acc, sub, -1)
}
return false
}

9
server/server.go
View File

@@ -3360,3 +3360,12 @@ func (s *Server) setFirstPingTimer(c *client) {
d += time.Duration(addDelay)
c.ping.tmr = time.AfterFunc(d, c.processPingTimer)
}
func (s *Server) updateRemoteSubscription(acc *Account, sub *subscription, delta int32) {
s.updateRouteSubscriptionMap(acc, sub, delta)
if s.gateway.enabled {
s.gatewayUpdateSubInterest(acc.Name, sub, delta)
}
s.updateLeafNodes(acc, sub, delta)
}