floraachy
/
console
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

sleep when no message was found in queue

This commit is contained in:
medcl 2023-05-26 12:02:39 +08:00
parent 14ed65fa94
commit 241c9f57fb
2 changed files with 9 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

4
plugin/elastic/activity.go
View File

@ -261,6 +261,10 @@ func (processor *ActivityProcessor) HandleMessage(ctx *pipeline.Context, qConfig
ctx1, messages, timeout, err := queue.Consume(qConfig, consumer, offset) ctx1, messages, timeout, err := queue.Consume(qConfig, consumer, offset)
if len(messages)==0{
time.Sleep(time.Millisecond * time.Duration(500))
}
if timeout { if timeout {
log.Tracef("timeout on queue:[%v]", qConfig.Name) log.Tracef("timeout on queue:[%v]", qConfig.Name)
ctx.Failed(fmt.Errorf("timeout on queue:[%v]", qConfig.Name)) ctx.Failed(fmt.Errorf("timeout on queue:[%v]", qConfig.Name))

5
plugin/elastic/metadata.go
View File

@ -285,6 +285,11 @@ func (processor *MetadataProcessor) HandleMessage(ctx *pipeline.Context, qConfig
return return
} }
ctx1, messages, isTimeout, err := queue.Consume(qConfig, consumer, offset) ctx1, messages, isTimeout, err := queue.Consume(qConfig, consumer, offset)
if len(messages)==0{
time.Sleep(time.Millisecond * time.Duration(500))
}
//if timeout{ //if timeout{
// log.Tracef("timeout on queue:[%v]",qConfig.Name) // log.Tracef("timeout on queue:[%v]",qConfig.Name)
// ctx.Failed() // ctx.Failed()