chtwrs/chirpnest
1
0
Fork 0
Code Issues Pull Requests Projects Releases Wiki Activity

client revamping race condition

Browse Source
This commit is contained in:
shoopea 2019-07-31 14:54:20 +08:00
parent d7bcdb0108
commit 74b6b141f0
1 changed files with 2 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

2
workers.go
View File

@ -571,6 +571,7 @@ func MQTidyKeepAliveWorker() {
t := time.Now() t := time.Now()
muxClients.Lock() muxClients.Lock()
for id, clt := range clients { for id, clt := range clients {
clt.Mux.Lock()
if clt.Active && clt.HeartBeat.Add(3*KeepAliveHeartBeatSeconds*time.Second).Before(time.Now()) { if clt.Active && clt.HeartBeat.Add(3*KeepAliveHeartBeatSeconds*time.Second).Before(time.Now()) {
msgs, err := clt.MQ.Channel.QueuePurge(clt.MQ.Queue.Name, false) msgs, err := clt.MQ.Channel.QueuePurge(clt.MQ.Queue.Name, false)
logOnError(err, "MQTidyKeepAliveWorker : Channel.QueuePurge()") logOnError(err, "MQTidyKeepAliveWorker : Channel.QueuePurge()")
@ -592,6 +593,7 @@ func MQTidyKeepAliveWorker() {
TGCmdQueue <- cmd TGCmdQueue <- cmd
clt.Active = false clt.Active = false
} }
clt.Mux.Unlock()
} }
muxClients.Unlock() muxClients.Unlock()
time.Sleep(time.Until(t.Add(time.Second))) time.Sleep(time.Until(t.Add(time.Second)))