| 380 | } |
| 381 | |
| 382 | func (b *NetworkTelemetryBatcher) Handler(events []*proto.TelemetryEvent) { |
| 383 | b.mu.Lock() |
| 384 | defer b.mu.Unlock() |
| 385 | select { |
| 386 | case <-b.closed: |
| 387 | return |
| 388 | default: |
| 389 | } |
| 390 | |
| 391 | for _, event := range events { |
| 392 | b.pending = append(b.pending, event) |
| 393 | |
| 394 | if len(b.pending) >= b.maxSize { |
| 395 | // This can't call sendTelemetryBatch directly because we already |
| 396 | // hold the lock. |
| 397 | events := b.pending |
| 398 | b.pending = []*proto.TelemetryEvent{} |
| 399 | // Resetting the ticker is best effort. We don't care if the ticker |
| 400 | // has already fired or has a pending message, because the only risk |
| 401 | // is that we send two telemetry events in short succession (which |
| 402 | // is totally fine). |
| 403 | b.ticker.Reset(b.frequency) |
| 404 | // Perform the send in a goroutine to avoid blocking the DRPC call. |
| 405 | go b.batchFn(events) |
| 406 | } |
| 407 | } |
| 408 | } |