from kafka import KafkaProducer
import json
import time


class ProducerServer(KafkaProducer):

    def __init__(self, input_file, topic, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file
        self.topic = topic

    #TODO we're generating a dummy data
    def generate_data(self):
        """
        Read input JSON file from disk and produce individual serialized rows to Kafka
        """
        print("generate_data - init")
        with open(self.input_file, "r") as f:

            # read JSON data from input file
            data = json.loads(f.read())

            for idx, row in enumerate(data):                
                # serialize Python dict to string
                msg = self.serialize_json(row)
                #print(f"Linha: {row}")
                self.send(self.topic, msg)
                self.flush()
                #print("Sleeping")
                time.sleep(1)

    # TODO fill this in to return the json dictionary to binary
    def dict_to_binary(self, json_dict):
        return 
        
        
    @staticmethod
    def serialize_json(json_data):
        """
        Serialize Python dict to JSON-formatted, UTF-8 encoded string
        """
        return json.dumps(json_data).encode("utf-8")        
d8f801cdbe5a30bf4861613ad88fbe90f167a15c
<mask token>


class ProducerServer(KafkaProducer):
    <mask token>
    <mask token>

    def dict_to_binary(self, json_dict):
        return
    <mask token>

<mask token>


class ProducerServer(KafkaProducer):

    def __init__(self, input_file, topic, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file
        self.topic = topic

    def generate_data(self):
        """
        Read input JSON file from disk and produce individual serialized rows to Kafka
        """
        print('generate_data - init')
        with open(self.input_file, 'r') as f:
            data = json.loads(f.read())
            for idx, row in enumerate(data):
                msg = self.serialize_json(row)
                self.send(self.topic, msg)
                self.flush()
                time.sleep(1)

    def dict_to_binary(self, json_dict):
        return
    <mask token>

<mask token>


class ProducerServer(KafkaProducer):

    def __init__(self, input_file, topic, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file
        self.topic = topic

    def generate_data(self):
        """
        Read input JSON file from disk and produce individual serialized rows to Kafka
        """
        print('generate_data - init')
        with open(self.input_file, 'r') as f:
            data = json.loads(f.read())
            for idx, row in enumerate(data):
                msg = self.serialize_json(row)
                self.send(self.topic, msg)
                self.flush()
                time.sleep(1)

    def dict_to_binary(self, json_dict):
        return

    @staticmethod
    def serialize_json(json_data):
        """
        Serialize Python dict to JSON-formatted, UTF-8 encoded string
        """
        return json.dumps(json_data).encode('utf-8')

from kafka import KafkaProducer
import json
import time


class ProducerServer(KafkaProducer):

    def __init__(self, input_file, topic, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file
        self.topic = topic

    def generate_data(self):
        """
        Read input JSON file from disk and produce individual serialized rows to Kafka
        """
        print('generate_data - init')
        with open(self.input_file, 'r') as f:
            data = json.loads(f.read())
            for idx, row in enumerate(data):
                msg = self.serialize_json(row)
                self.send(self.topic, msg)
                self.flush()
                time.sleep(1)

    def dict_to_binary(self, json_dict):
        return

    @staticmethod
    def serialize_json(json_data):
        """
        Serialize Python dict to JSON-formatted, UTF-8 encoded string
        """
        return json.dumps(json_data).encode('utf-8')

from kafka import KafkaProducer
import json
import time


class ProducerServer(KafkaProducer):

    def __init__(self, input_file, topic, **kwargs):
        super().__init__(**kwargs)
        self.input_file = input_file
        self.topic = topic

    #TODO we're generating a dummy data
    def generate_data(self):
        """
        Read input JSON file from disk and produce individual serialized rows to Kafka
        """
        print("generate_data - init")
        with open(self.input_file, "r") as f:

            # read JSON data from input file
            data = json.loads(f.read())

            for idx, row in enumerate(data):                
                # serialize Python dict to string
                msg = self.serialize_json(row)
                #print(f"Linha: {row}")
                self.send(self.topic, msg)
                self.flush()
                #print("Sleeping")
                time.sleep(1)

    # TODO fill this in to return the json dictionary to binary
    def dict_to_binary(self, json_dict):
        return 
        
        
    @staticmethod
    def serialize_json(json_data):
        """
        Serialize Python dict to JSON-formatted, UTF-8 encoded string
        """
        return json.dumps(json_data).encode("utf-8")        