otel-data-replay/oteldatareplay/record_reader.py

import json
import logging
from typing import Generator

from oteldatareplay.resource_span_record import ResourceSpanRecord

logger = logging.getLogger(__name__)


class RecordReader(object):
    def __init__(self, file_path):
        self.file_path = file_path
        self.fp = open(file_path, 'r')
        self.running = True

    def reset(self):
        self.fp.seek(0)

    def records_generator(self) -> Generator[ResourceSpanRecord, None, None]:
        line = self.fp.readline()
        while line and self.running:
            ob = json.loads(line)
            yield ResourceSpanRecord(ob)
            line = self.fp.readline()
        return

    def close(self):
        self.running = False
        self.fp.close()
        logger.info("Player closed")