43 lines
1.1 KiB
Python
43 lines
1.1 KiB
Python
from flask import Flask, request
|
|
import json
|
|
import redis
|
|
import mysql.connector
|
|
from confluent_kafka import Producer, Consumer, KafkaError
|
|
|
|
app = Flask(__name__)
|
|
cache = redis.Redis(host='redis-service', port=6379)
|
|
|
|
# Kafka producer configuration
|
|
p = Producer({'bootstrap.servers': 'my-kafka-service:9092'}) # replace 'my-kafka-service' with your Kafka service name
|
|
|
|
# Kafka consumer configuration
|
|
c = Consumer({
|
|
'bootstrap.servers': 'my-kafka-service:9092', # replace 'my-kafka-service' with your Kafka service name
|
|
'group.id': 'mygroup',
|
|
'auto.offset.reset': 'earliest'
|
|
})
|
|
|
|
c.subscribe(['mytopic'])
|
|
|
|
@app.route('/produce', methods=['POST'])
|
|
def produce():
|
|
data = request.json
|
|
p.produce('mytopic', json.dumps(data))
|
|
return 'Message produced'
|
|
|
|
@app.route('/consume')
|
|
def consume():
|
|
msg = c.poll(1.0)
|
|
|
|
if msg is None:
|
|
return 'No message'
|
|
if msg.error():
|
|
return 'Error: {}'.format(msg.error())
|
|
|
|
data = json.loads(msg.value().decode('utf-8'))
|
|
return data
|
|
|
|
if __name__ == '__main__':
|
|
app.run(host='0.0.0.0', port=80)
|
|
|