SendMessage.py

import datetime
import json
from kafka import KafkaProducer,KafkaConsumer

# import pdb; pdb.set_trace() #python  debugger


class SendingMessage:

    
    def one_to_one(self,sender_id,reciever_id,msg,timestamp,msg_type):
        
        details={
            "Sender":sender_id,
            "Reciever":reciever_id,
            "MessageContent":msg,
            "Timestamp":timestamp,
            "MessageType":msg_type
        }

        return details
    


time=datetime.datetime.now()
user1=SendingMessage()
message=user1.one_to_one(2456,178,"Hi, How are you?,This is Sanjay,How are you?",time.strftime("%c"),"String")

producer=KafkaProducer(bootstrap_servers='localhost:9092',value_serializer=lambda v:json.dumps(v).encode('utf-8'))
producer.send('onetoone',value=message)
producer.flush()

print("Message sent")


consumer=KafkaConsumer('onetoone',bootstrap_servers='localhost:9092',auto_offset_reset='earliest')

for message in consumer:
    print(f"Recieved Message:{message.value}")


consumer.close()