import time
import asyncio
from aiokafka import AIOKafkaProducer

from settings import KAFKA_SERVERS, SAVEPOINT, LOG_FILE, KAFKA_TOPIC


class LogStreamer:
    def __init__(self,
                 KAFKA_SERVERS,
                 KAFKA_TOPIC,
                 loop,
                 savepoint_file,
                 log_file):
        


    async def produce(self, finite=False):
        last = self.savepoint_file.read()
        if last:
            self.log_file.seek(int(last))

        skip_first_empty = True
        while True:
            line = self.log_file.readline()
            line = line.strip(' \t\n\r')
            if not line:
                if finite and not skip_first_empty:
                    return

                skip_first_empty = False
                time.sleep(0.1)

                current_position = self.log_file.tell()

                if last != current_position:
                    self.savepoint_file.seek(0)
                    self.savepoint_file.write(str(current_position))

                continue

            '''
            Here we can convert our data to JSON. But I because JSON performance is not extremely good
            with standart libraries, and because we use asynchronous non-blocking model here, I think it's
            best to just pass data as is. I want to create as little as possible overhead here. We want to
            stream data as fast as possible.
            '''

            await self.producer.send_and_wait(self.KAFKA_TOPIC, line.encode())

    def start(self):
        self.loop.run_until_complete(self.producer.start())
        self.loop.run_until_complete(self.produce())
        self.loop.run_until_complete(self.producer.stop())
        self.loop.close()

if __name__ == '__main__':
    with open(SAVEPOINT, 'r+') as savepoint_file, open(LOG_FILE, 'r') as log_file:
        streamer = LogStreamer(KAFKA_SERVERS,
                               KAFKA_TOPIC,
                               asyncio.get_event_loop(),
                               savepoint_file,
                               log_file)
        streamer.start()

self.KAFKA_TOPIC = KAFKA_TOPIC
        self.loop = loop
        self.producer = AIOKafkaProducer(loop=self.loop, bootstrap_servers=KAFKA_SERVERS)
        self.savepoint_file = savepoint_file
        self.log_file = log_file