refreshQueue 同步产品线的报警队列
(products []*types.Product)
| 84 | |
| 85 | //refreshQueue 同步产品线的报警队列 |
| 86 | func (c *Controller) refreshQueue(products []*types.Product) { |
| 87 | c.eventQueuesMutex.Lock() |
| 88 | defer c.eventQueuesMutex.Unlock() |
| 89 | now := time.Now() |
| 90 | for _, product := range products { |
| 91 | if _, ok := c.eventQueues[product.ID]; !ok { |
| 92 | lg.Info("create event queue %s", product.Name) |
| 93 | eventPool := NewEventPool(product.Name, GlobalConfig.EVENT_POOL_SIZE) |
| 94 | c.eventQueues[product.ID] = eventPool |
| 95 | go processSingleQueue(eventPool) |
| 96 | } |
| 97 | lg.Info("refresh event queue %s", product.Name) |
| 98 | c.eventQueues[product.ID].update_time = now |
| 99 | } |
| 100 | for p, q := range c.eventQueues { |
| 101 | if now.Sub(q.update_time).Minutes() > 10 { |
| 102 | lg.Warn("product id %s event queue update time is expires, delete", p) |
| 103 | delete(c.eventQueues, p) |
| 104 | } |
| 105 | } |
| 106 | } |
no test coverage detected