| 372 | } |
| 373 | |
| 374 | func (da *Auditor) flushWriteQueue(timeout time.Duration) bool { |
| 375 | da.mu.RLock() |
| 376 | ch := da.writeCh |
| 377 | da.mu.RUnlock() |
| 378 | if ch == nil { |
| 379 | return true |
| 380 | } |
| 381 | |
| 382 | ack := make(chan struct{}) |
| 383 | task := writeTask{flush: ack} |
| 384 | timer := time.NewTimer(timeout) |
| 385 | defer timer.Stop() |
| 386 | |
| 387 | select { |
| 388 | case ch <- task: |
| 389 | case <-timer.C: |
| 390 | da.recordWriteError(fmt.Errorf("flush write queue: timed out while enqueueing sentinel")) |
| 391 | return false |
| 392 | } |
| 393 | |
| 394 | timer.Reset(timeout) |
| 395 | select { |
| 396 | case <-ack: |
| 397 | return true |
| 398 | case <-timer.C: |
| 399 | da.recordWriteError(fmt.Errorf("flush write queue: timed out waiting for sentinel")) |
| 400 | return false |
| 401 | } |
| 402 | } |
| 403 | |
| 404 | func (da *Auditor) recordWriteError(err error) { |
| 405 | if err == nil { |