package publicfault
import (
"context"
"sync"
"time"
"clusterd/pkg/application/statistics"
"clusterd/pkg/domain/publicfault"
)
var PubFaultNeedDelete *needDeleteQueue
func init() {
PubFaultNeedDelete = &needDeleteQueue{
faults: make([]needDeleteFault, 0),
mutex: sync.Mutex{},
}
}
type needDeleteQueue struct {
faults []needDeleteFault
mutex sync.Mutex
}
type needDeleteFault struct {
deleteTime int64
nodeName string
faultKey string
}
func (q *needDeleteQueue) Push(deleteTime int64, nodeName, faultKey string) {
newItem := needDeleteFault{
deleteTime: deleteTime,
nodeName: nodeName,
faultKey: faultKey,
}
q.mutex.Lock()
defer q.mutex.Unlock()
q.faults = append(q.faults, newItem)
}
func (q *needDeleteQueue) Pop() needDeleteFault {
q.mutex.Lock()
defer q.mutex.Unlock()
if len(q.faults) == 0 {
return needDeleteFault{}
}
removed := q.faults[0]
q.faults = q.faults[1:]
return removed
}
func (q *needDeleteQueue) Len() int {
q.mutex.Lock()
defer q.mutex.Unlock()
return len(q.faults)
}
func (q *needDeleteQueue) DealDelete(ctx context.Context) {
const duration = 500 * time.Millisecond
for {
select {
case <-ctx.Done():
return
default:
if q.Len() == 0 {
time.Sleep(duration)
continue
}
needDeal := q.Pop()
deleteTime := needDeal.deleteTime
if deleteTime <= time.Now().Unix() {
publicfault.PubFaultCache.DeleteOccurFault(needDeal.nodeName, needDeal.faultKey)
statistics.StatisticFault.Notify()
continue
}
diffTime := (deleteTime - time.Now().Unix()) * int64(time.Second)
time.Sleep(time.Duration(diffTime))
publicfault.PubFaultCache.DeleteOccurFault(needDeal.nodeName, needDeal.faultKey)
statistics.StatisticFault.Notify()
}
}
}