@@ -20,14 +20,16 @@ package event
2020import (
2121 "fmt"
2222 "net/http"
23- "sync"
2423 "time"
2524
25+ "golang.org/x/sync/semaphore"
26+
2627 "github.com/optimizely/go-sdk/pkg/logging"
2728 "github.com/optimizely/go-sdk/pkg/metrics"
2829 "github.com/optimizely/go-sdk/pkg/utils"
2930)
3031
32+ const maxWorkers = int64 (1 )
3133const maxRetries = 3
3234const defaultQueueSize = 1000
3335const sleepTime = 1 * time .Second
@@ -40,7 +42,7 @@ type Dispatcher interface {
4042// httpEventDispatcher is the HTTP implementation of the Dispatcher interface
4143type httpEventDispatcher struct {
4244 requester * utils.HTTPRequester
43- logger logging.OptimizelyLogProducer
45+ logger logging.OptimizelyLogProducer
4446}
4547
4648// DispatchEvent dispatches event with callback
@@ -79,39 +81,38 @@ func NewHTTPEventDispatcher(sdkKey string, requester *utils.HTTPRequester, logge
7981
8082// QueueEventDispatcher is a queued version of the event Dispatcher that queues, returns success, and dispatches events in the background
8183type QueueEventDispatcher struct {
82- eventQueue Queue
83- eventFlushLock sync. Mutex
84- Dispatcher Dispatcher
85- logger logging.OptimizelyLogProducer
84+ eventQueue Queue
85+ processing * semaphore. Weighted
86+ Dispatcher Dispatcher
87+ logger logging.OptimizelyLogProducer
8688
8789 // metrics
88- queueSize metrics.Gauge
89- sucessFlush metrics.Counter
90- failFlushCounter metrics.Counter
91- retryFlushCounter metrics.Counter
90+ queueSizeGauge metrics.Gauge
91+ sucessFlushCounter metrics.Counter
92+ failFlushCounter metrics.Counter
93+ retryFlushCounter metrics.Counter
9294}
9395
9496// DispatchEvent queues event with callback and calls flush in a go routine.
9597func (ed * QueueEventDispatcher ) DispatchEvent (event LogEvent ) (bool , error ) {
9698 ed .eventQueue .Add (event )
97- go func () {
98- ed .flushEvents ()
99- }()
99+ go ed .flushEvents ()
100100 return true , nil
101101}
102102
103103// flush the events
104104func (ed * QueueEventDispatcher ) flushEvents () {
105105
106- ed . eventFlushLock . Lock ()
107-
108- defer func () {
109- ed . eventFlushLock . Unlock ()
110- }( )
106+ // Limit flushing to a single worker
107+ if ! ed . processing . TryAcquire ( 1 ) {
108+ return
109+ }
110+ defer ed . processing . Release ( 1 )
111111
112112 retryCount := 0
113- ed .queueSize .Set (float64 (ed .eventQueue .Size ()))
114- for ed .eventQueue .Size () > 0 {
113+ queueSize := ed .eventQueue .Size ()
114+ for ; queueSize > 0 ; queueSize = ed .eventQueue .Size () {
115+ ed .queueSizeGauge .Set (float64 (queueSize ))
115116 if retryCount > maxRetries {
116117 ed .logger .Error (fmt .Sprintf ("event failed to send %d times. It will retry on next event sent" , maxRetries ), nil )
117118 ed .failFlushCounter .Add (1 )
@@ -136,10 +137,10 @@ func (ed *QueueEventDispatcher) flushEvents() {
136137
137138 if err == nil {
138139 if success {
139- ed .logger .Debug (fmt . Sprintf ( "Dispatched log event %+v" , event ) )
140+ ed .logger .Debug ("dispatch log event succeeded"
140141 ed .eventQueue .Remove (1 )
141142 retryCount = 0
142- ed .sucessFlush .Add (1 )
143+ ed .sucessFlushCounter .Add (1 )
143144 } else {
144145 ed .logger .Warning ("dispatch event failed" )
145146 // we failed. Sleep some seconds and try again.
@@ -159,7 +160,7 @@ func (ed *QueueEventDispatcher) flushEvents() {
159160 ed .retryFlushCounter .Add (1 )
160161 }
161162 }
162- ed .queueSize .Set (float64 (ed . eventQueue . Size () ))
163+ ed .queueSizeGauge .Set (float64 (queueSize ))
163164}
164165
165166// NewQueueEventDispatcher creates a Dispatcher that queues in memory and then sends via go routine.
@@ -172,13 +173,15 @@ func NewQueueEventDispatcher(sdkKey string, metricsRegistry metrics.Registry) *Q
172173 dispatcherMetricsRegistry = metrics .NewNoopRegistry () // protective code to set
173174 }
174175
176+ logger := logging .GetLogger (sdkKey , "QueueEventDispatcher" )
175177 return & QueueEventDispatcher {
176- eventQueue : NewInMemoryQueue (defaultQueueSize ),
177- Dispatcher : NewHTTPEventDispatcher (sdkKey , nil , nil ),
178- queueSize : dispatcherMetricsRegistry .GetGauge (metrics .DispatcherQueueSize ),
179- retryFlushCounter : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherRetryFlush ),
180- failFlushCounter : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherFailedFlush ),
181- sucessFlush : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherSuccessFlush ),
182- logger : logging .GetLogger (sdkKey , "QueueEventDispatcher" ),
178+ eventQueue : NewInMemoryQueueWithLogger (defaultQueueSize , logger ),
179+ Dispatcher : NewHTTPEventDispatcher (sdkKey , nil , nil ),
180+ queueSizeGauge : dispatcherMetricsRegistry .GetGauge (metrics .DispatcherQueueSize ),
181+ retryFlushCounter : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherRetryFlush ),
182+ failFlushCounter : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherFailedFlush ),
183+ sucessFlushCounter : dispatcherMetricsRegistry .GetCounter (metrics .DispatcherSuccessFlush ),
184+ logger : logger ,
185+ processing : semaphore .NewWeighted (maxWorkers ),
183186 }
184187}
0 commit comments