(fix): cleanup event processor #167
Conversation
pkg/event/processor.go
Outdated
|
|
||
| p.Q.Add(event) | ||
|
|
||
| if p.Q.Size() >= p.BatchSize { |
There was a problem hiding this comment.
Can we flatten this out:
if p.Q.Size() < p.BatchSize || p.processing {
return
}
p.processing = true
go func() {
p.FlushEvents()
p.processing = false
}()
pkg/event/processor.go
Outdated
| @@ -44,6 +44,7 @@ type BatchEventProcessor struct { | |||
| Mux sync.Mutex | |||
| Ticker *time.Ticker | |||
| EventDispatcher Dispatcher | |||
| processing bool | |||
There was a problem hiding this comment.
This is like a poor man's Mutex :)
There was a problem hiding this comment.
you noticed that. Yeah, I set it and test it in the same thread and then set it back in the other thread. So, there is really no chance of collision. However, go lint disagreed. So, I went ahead and created a new util struct called AtomicProperty.
pkg/event/processor.go
Outdated
| p.Q.Add(event) | ||
|
|
||
| if p.Q.Size() >= p.BatchSize { | ||
| if processing, ok := p.processing.Get().(bool); ok && !processing { |
There was a problem hiding this comment.
Perhaps you can use a semaphore.
if p.sem.TryAquire(1) {
go func() {
p.FlushEvents()
p.sem.Release(1)
}()
}ref: https://medium.com/@deckarep/gos-extended-concurrency-semaphores-part-1-5eeabfa351ce
| @@ -123,6 +128,11 @@ func NewBatchEventProcessor(options ...BPOptionConfig) *BatchEventProcessor { | |||
| p.BatchSize = DefaultBatchSize | |||
| } | |||
|
|
|||
| if p.BatchSize > p.MaxQueueSize { | |||
| pLogger.Warning("Batch size is larger than queue size. Swapping") | |||
| p.BatchSize, p.MaxQueueSize = p.MaxQueueSize, p.BatchSize | |||
There was a problem hiding this comment.
This is a bit sneaky and I'm not sure we want to swap parameters like this.
There was a problem hiding this comment.
actually, it was a lint suggestion. i love it! :)
Summary
Testing
This PR has unit tests for all the fixes above.