153 lines
5.2 KiB
Python
153 lines
5.2 KiB
Python
from confluent_kafka import Consumer, Producer
|
|
import json
|
|
from app.utils.logger import logger
|
|
from app.config import settings
|
|
import requests
|
|
from app.integrations.simbrella import SimbrellaClient
|
|
|
|
|
|
class KafkaIntegration:
|
|
|
|
BASE_URL = settings.BANK_CALL_BASE_URL
|
|
|
|
_consumer = None
|
|
|
|
_consumer_config = {
|
|
"bootstrap.servers": settings.KAFKA_BROKER,
|
|
"group.id": "loan-service-consumer",
|
|
"auto.offset.reset": "earliest",
|
|
"enable.auto.commit": True,
|
|
}
|
|
|
|
@staticmethod
|
|
def _get_consumer():
|
|
"""Kafka consumer"""
|
|
if not KafkaIntegration._consumer:
|
|
KafkaIntegration._consumer = Consumer(KafkaIntegration._consumer_config)
|
|
logger.info(
|
|
f"Consumer connected to Kafka broker at {KafkaIntegration._consumer_config['bootstrap.servers']}"
|
|
)
|
|
return KafkaIntegration._consumer
|
|
|
|
@staticmethod
|
|
def delivery_report(err, msg):
|
|
"""Called once for each message produced"""
|
|
if err is not None:
|
|
logger.error(f"Message delivery failed: {err}")
|
|
raise RuntimeError(f"Message delivery failed: {err}")
|
|
|
|
else:
|
|
logger.debug(
|
|
f"Message delivered to {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}"
|
|
)
|
|
|
|
@staticmethod
|
|
def receive_disbursement_messages(topic, timeout):
|
|
"""
|
|
Receive messages from a Kafka topic.
|
|
|
|
:param topic: The Kafka topic to subscribe to
|
|
:param timeout: Time to wait for a message (in seconds)
|
|
:return: The message value (decoded) or None if no message is received
|
|
"""
|
|
consumer = KafkaIntegration._get_consumer()
|
|
# consumer.subscribe(["PROCESS_PAYMENT"])
|
|
# consumer.subscribe([topic])
|
|
consumer.subscribe(['LOAN_REPAYMENT', 'PROCESS_PAYMENT'])
|
|
|
|
|
|
logger.info(
|
|
f"Waiting for messages from topic {topic} with this timeout: {timeout}..."
|
|
)
|
|
message = []
|
|
try:
|
|
msg = consumer.poll(timeout=timeout)
|
|
|
|
logger.info(str(msg.value))
|
|
|
|
logger.info(
|
|
f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: "
|
|
)
|
|
|
|
if msg is None:
|
|
logger.info(f"No message received from topic {topic} within timeout")
|
|
return None
|
|
if msg.error():
|
|
logger.info(f"Consumer error: {msg.error()}")
|
|
raise RuntimeError(f"Consumer error: {msg.error()}")
|
|
|
|
# Decode and return the message value
|
|
message_value= {"value":''}
|
|
if msg.value() != "":
|
|
message_value = msg.value().decode("utf-8")
|
|
logger.info(
|
|
f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message_value}"
|
|
)
|
|
|
|
# Invalid Json will stop this process
|
|
try:
|
|
message = json.loads(message_value) if message_value else None
|
|
except ValueError as e:
|
|
return message # Invalid JSON JUST TURN BACK HERE
|
|
else:
|
|
pass # valid json
|
|
|
|
# Call the endpoint if provided
|
|
if message:
|
|
logger.info(
|
|
f"Received message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message}"
|
|
)
|
|
current_topic = msg.topic()
|
|
|
|
if current_topic=="PROCESS_PAYMENT":
|
|
KafkaIntegration._call_disbursement_service(message)
|
|
|
|
if current_topic=="LOAN_REPAYMENT":
|
|
KafkaIntegration._call_collect_loan_service(message)
|
|
logger.info(
|
|
f"Loan Repayment message from {msg.topic()} [{msg.partition()}] @ offset {msg.offset()}: {message}"
|
|
)
|
|
|
|
return message
|
|
|
|
except Exception as e:
|
|
logger.error(f"Error while receiving message: {e}")
|
|
raise
|
|
|
|
#return []
|
|
|
|
@staticmethod
|
|
def close_consumer():
|
|
"""Shutdown consumer"""
|
|
consumer = KafkaIntegration._get_consumer()
|
|
consumer.close()
|
|
logger.info("Kafka consumer closed")
|
|
|
|
@staticmethod
|
|
def _call_disbursement_service(message):
|
|
"""Call the disbursement service with the received message"""
|
|
logger.info(f"Calling disbursement service with message: {message}")
|
|
|
|
try:
|
|
response = SimbrellaClient.disbursement(message)
|
|
logger.info(
|
|
f"Successfully sent message to disbursement service: {response.status_code}"
|
|
)
|
|
except Exception as e:
|
|
logger.info(f"Failed to call disbursement service: {e}")
|
|
#raise
|
|
|
|
@staticmethod
|
|
def _call_collect_loan_service(message):
|
|
"""Call the collect loan service with the received message"""
|
|
logger.info(f"Calling collect_loan service with message: {message}")
|
|
|
|
try:
|
|
response = SimbrellaClient.collect_loan(message)
|
|
logger.info(
|
|
f"Successfully sent message to collect_loan service: {response.status_code}"
|
|
)
|
|
except Exception as e:
|
|
logger.info(f"Failed to call collect_loan service: {e}")
|
|
# raise
|