kafka consumer to ruby on rails

I am currently working with a rails server which is supposed to run python script, which are kafka consumer/producer.

The server must to run the script then receive the processed data from consumer, and render them to the site.

I am able to run a script but can not fund a solution for the consumer to be connected. As the consumer is either running non stop accepting the messages, or running in while loop. I tried to run the consumer first from ruby, which starts the consumer, but never gets the consumer, as it is listening, but the other script could not be run.

So the flow of the message ideally should be something like this -> email from logged user to kafka producer -> MQ -> kafka consumer generates data writes to db -> producer query data from database -> MQ -> consumer accepts the data and renders them to the site.

The ideal scenario would be a one script lets call it manager that does all the work only accepts data and returns it. It also was not able to do that because, the one script also runs consumer and listens for producer, but it is never ran.

so here is my code:

from kafka import KafkaProducer
from faker import Faker
import json
import time
class producer1():

        '''
        fr_instance= Faker()
        def get_new_user():
        return {"email_address":fr_instance.email(),"first_name": fr_instance.first_name(),
        "lastname": fr_instance.last_name(), "occupation":fr_instance.job()}
        '''
        def __init__(self):
                self
                

        def json_serializer(self, data):
                return json.dumps(data).encode("utf-8")



        def send(self,email):
                print(email)
                producer = KafkaProducer(bootstrap_servers='localhost:9092',
                        value_serializer = self.json_serializer)
                registred_user = {"email": email}
                future = producer.send("NewUserTopic", registred_user)
                print (registred_user)
                result = future.get(timeout=10)
                

p = producer1()
if __name__ == '__main__':   
        email = "[email protected]"
        p.send(email)
    

then 1st consumer:

from kafka import KafkaConsumer
import json
import random
from sqlalchemy.orm import sessionmaker
import dbservice
import time

class consumer1():
    def __init__(self) -> None:
        self
    def email(self):
        consumer = KafkaConsumer('NewUserTopic',
        bootstrap_servers='localhost:9092',
        auto_offset_reset = 'latest', enable_auto_commit= False)
        for msg in consumer:
            
            
            msg_out = json.loads(msg.value)
            for value in  msg_out.values():
                

            
            #return print(msg_out)
                return (value)
            
    #generate dummy address , eth
    def gen_ETHw (self):
         numbers = str(random.randint(11111,99999))
         wallet_num = str("Ox"+numbers)
         return (wallet_num)
    #generate dummy address , btc
    def gen_BTCw (self):

         numbers = str(random.randint(11111,99999))
         wallet_num = str("Ox"+numbers)
         return (wallet_num)
    
    def commit_db (self, email, ETHw, BTCw):
        Session = sessionmaker(bind=dbservice.engine)
        s = Session()
        
        input = dbservice.walletdb( email,ETHw, BTCw)
        time.sleep(2)
        s.add(input)
        s.commit()
    

         
if __name__ =='__main__':
    while True:
        c = consumer1()
        c.commit_db(c.email(),c.gen_ETHw(),c.gen_BTCw())

query producer:

import dbservice
import dbservice
from sqlalchemy.orm import sessionmaker
from kafka import KafkaProducer
import json


class query_prod ():
        def __init__(self, email) -> None:
                self = self
                self.email = email
               
        
        def json_serializer(data):
                return json.dumps(data).encode("utf-8")

        producer = KafkaProducer(bootstrap_servers='localhost:9092',
                        value_serializer = json_serializer)

        Session = sessionmaker(bind=dbservice.engine)
        s = Session()
        
        def query_address(self,email):
                Session = sessionmaker(bind=dbservice.engine)
                s = Session()
                for s in s.query(dbservice.walletdb).filter_by(email=email):
                       return {"email":s.email,"ETH_w":s.ETH_w,"BTC_w":s.BTC_w}

        def send(self, email):
                data_to_send = self.query_address(email)
        
                future = self.producer.send("QueryAdressToServer", data_to_send)
                print (data_to_send)
                result = future.get(timeout=10)

if __name__ == '__main__':
        email = "[email protected]"
        query_prod=query_prod(email)
        query_prod.send(email)
    

and consume data which should be returned to the site:

from kafka import KafkaConsumer
import json
import time

class consume_for_web():
    string=""
    def __init__(self) -> None:
        self = self
        string = self.string
    
    
    def consumer(self):
        consumer = KafkaConsumer('QueryAdressToServer',
            bootstrap_servers='localhost:9092',
            auto_offset_reset = 'latest', enable_auto_commit= False)
        print('starting consumer')
        for msg in consumer:
            data = (('{}'.format(json.loads(msg.value))))
            self.string = self.string + data
            return print(data)
        
    def read_str(self):
        return print(self.string)

if __name__ =='__main__':
    while True:
        c = consume_for_web()
        c.consumer()
        ##print("reading")
        #c.read_str()

and finally my rails pages controller:

class PagesController < ApplicationController
  def home
  end

  def about
  end

  before_action :require_login
  def genw
    our_input = current_user.email
    puts our_input
    
    @consumer_result = `python3 /Users/samuelrybar/python_projects/Kafka_demo1/kafka-prod-coms/consumer2.py`

  end

  
  def mywa
  end
  def save
    
  end
end

Thanks for your time and help, I really appreciate it. :))