backgroundSend is the worker goroutine for sending events.
()
| 240 | |
| 241 | // backgroundSend is the worker goroutine for sending events. |
| 242 | func (oq *destinationQueue) backgroundSend() { |
| 243 | // Check if a worker is already running, and if it isn't, then |
| 244 | // mark it as started. |
| 245 | if !oq.running.CAS(false, true) { |
| 246 | return |
| 247 | } |
| 248 | destinationQueueRunning.Inc() |
| 249 | defer destinationQueueRunning.Dec() |
| 250 | defer oq.queues.clearQueue(oq) |
| 251 | defer oq.running.Store(false) |
| 252 | |
| 253 | // Mark the queue as overflowed, so we will consult the database |
| 254 | // to see if there's anything new to send. |
| 255 | oq.overflowed.Store(true) |
| 256 | |
| 257 | for { |
| 258 | // If we are overflowing memory and have sent things out to the |
| 259 | // database then we can look up what those things are. |
| 260 | if oq.overflowed.Load() { |
| 261 | oq.getPendingFromDatabase() |
| 262 | } |
| 263 | |
| 264 | // If we have nothing to do then wait either for incoming events, or |
| 265 | // until we hit an idle timeout. |
| 266 | select { |
| 267 | case <-oq.notify: |
| 268 | // There's work to do, either because getPendingFromDatabase |
| 269 | // told us there is, or because a new event has come in via |
| 270 | // sendEvent/sendEDU. |
| 271 | case <-time.After(queueIdleTimeout): |
| 272 | // The worker is idle so stop the goroutine. It'll get |
| 273 | // restarted automatically the next time we have an event to |
| 274 | // send. |
| 275 | return |
| 276 | case <-oq.process.Context().Done(): |
| 277 | // The parent process is shutting down, so stop. |
| 278 | return |
| 279 | } |
| 280 | |
| 281 | // If we are backing off this server then wait for the |
| 282 | // backoff duration to complete first, or until explicitly |
| 283 | // told to retry. |
| 284 | until, blacklisted := oq.statistics.BackoffInfo() |
| 285 | if blacklisted { |
| 286 | // It's been suggested that we should give up because the backoff |
| 287 | // has exceeded a maximum allowable value. Clean up the in-memory |
| 288 | // buffers at this point. The PDU clean-up is already on a defer. |
| 289 | logrus.Warnf("Blacklisting %q due to exceeding backoff threshold", oq.destination) |
| 290 | oq.pendingMutex.Lock() |
| 291 | for i := range oq.pendingPDUs { |
| 292 | oq.pendingPDUs[i] = nil |
| 293 | } |
| 294 | for i := range oq.pendingEDUs { |
| 295 | oq.pendingEDUs[i] = nil |
| 296 | } |
| 297 | oq.pendingPDUs = nil |
| 298 | oq.pendingEDUs = nil |
| 299 | oq.pendingMutex.Unlock() |
no test coverage detected