def run_once(self):
"""Accumulate records and flush when accumulator is ready."""
try:
record = self.queue.get(timeout=0.05)
except queue.Empty:
record = None
else:
success = self._accumulator.try_append(record)
if not success:
self.flush()
success = self._accumulator.try_append(record)
assert success, "Failed to accumulate even after flushing"
self.queue.task_done()
is_ready = self._accumulator.is_ready()
force_flush = not self._running and record is None
if is_ready or force_flush:
self.flush()
评论列表
文章目录