Merge pull request #209 from xmidt-org/remove-queueEmpty

removed queueEmpty variable from outbound sender

CHANGELOG.md

@@ -7,6 +7,7 @@ and this project adheres to [Semantic Versioning](http://semver.org/spec/v2.0.0.
 ## [Unreleased]
 - cleaned up shutdown logic for outbound sender [#205](https://github.com/xmidt-org/caduceus/pull/205)
 - added resetting queue depth and current workers gauges to outbound sender [#205](https://github.com/xmidt-org/caduceus/pull/205)
+- removed queueEmpty variable from outbound sender [#209](https://github.com/xmidt-org/caduceus/pull/209)
 
 ## [v0.2.7]
 - pared down logging, especially debugging logs [#196](https://github.com/xmidt-org/caduceus/pull/196)

outboundSender.go

@@ -154,7 +154,6 @@ type CaduceusOutboundSender struct {
 	failureMsg                       FailureMessage
 	logger                           log.Logger
 	mutex                            sync.RWMutex
-	queueEmpty                       bool
 	queue                            atomic.Value
 }
 
@@ -197,7 +196,6 @@ func (osf OutboundSenderFactory) New() (obs OutboundSender, err error) {
 			QueueSize:    osf.QueueSize,
 			Workers:      osf.NumWorkers,
 		},
-		queueEmpty: true,
 	}
 
 	// Don't share the secret with others when there is an error.
@@ -427,7 +425,7 @@ func (obs *CaduceusOutboundSender) Queue(msg *wrp.Message) {
 }
 
 func (obs *CaduceusOutboundSender) isValidTimeWindow(now, dropUntil, deliverUntil time.Time) bool {
-	if false == now.After(dropUntil) || !obs.queueEmpty {
+	if false == now.After(dropUntil) {
 		// client was cut off
 		obs.droppedCutoffCounter.Add(1.0)
 		return false
@@ -447,8 +445,6 @@ func (obs *CaduceusOutboundSender) Empty(droppedCounter metrics.Counter) {
 	obs.queue.Store(make(chan *wrp.Message, obs.queueSize))
 	droppedCounter.Add(float64(len(droppedMsgs)))
 	obs.queueDepthGauge.Set(0.0)
-	obs.queueEmpty = true
-
 	return
 }
 
@@ -620,7 +616,6 @@ func (obs *CaduceusOutboundSender) queueOverflow() {
 		obs.mutex.Unlock()
 		return
 	}
-	obs.queueEmpty = false
 	obs.dropUntil = time.Now().Add(obs.cutOffPeriod)
 	obs.dropUntilGauge.Set(float64(obs.dropUntil.Unix()))
 	secret := obs.listener.Config.Secret