1
0
mirror of https://github.com/daniviga/bite.git synced 2024-11-22 21:16:12 +01:00

Add kafka settings

This commit is contained in:
Daniele Viganò 2023-09-08 18:18:42 +02:00
parent 49211437d2
commit e3785d4669
3 changed files with 16 additions and 5 deletions

View File

@ -158,6 +158,11 @@ MQTT_BROKER = {
'PORT': '1883', 'PORT': '1883',
} }
KAFKA_BROKER = {
'HOST': 'kafka',
'PORT': '9092',
}
# If no local_settings.py is availble in the current folder let's try to # If no local_settings.py is availble in the current folder let's try to
# load it from the application root # load it from the application root
try: try:

View File

@ -39,6 +39,8 @@ class Command(BaseCommand):
MQTT_HOST = settings.MQTT_BROKER['HOST'] MQTT_HOST = settings.MQTT_BROKER['HOST']
MQTT_PORT = int(settings.MQTT_BROKER['PORT']) MQTT_PORT = int(settings.MQTT_BROKER['PORT'])
KAFKA_HOST = settings.KAFKA_BROKER['HOST']
KAFKA_PORT = int(settings.KAFKA_BROKER['PORT'])
producer = None producer = None
@sync_to_async @sync_to_async
@ -85,7 +87,9 @@ class Command(BaseCommand):
while True: while True:
try: try:
self.producer = KafkaProducer( self.producer = KafkaProducer(
bootstrap_servers='localhost:9092', bootstrap_servers='{}:{}'.format(
self.KAFKA_HOST, self.KAFKA_PORT
),
value_serializer=lambda v: json.dumps(v).encode('utf-8'), value_serializer=lambda v: json.dumps(v).encode('utf-8'),
retries=5 retries=5
) )

View File

@ -29,13 +29,13 @@ from django.core.exceptions import ObjectDoesNotExist
from api.models import Device from api.models import Device
from telemetry.models import Telemetry from telemetry.models import Telemetry
MQTT_HOST = settings.MQTT_BROKER['HOST']
MQTT_PORT = int(settings.MQTT_BROKER['PORT'])
class Command(BaseCommand): class Command(BaseCommand):
help = 'MQTT to DB deamon' help = 'MQTT to DB deamon'
KAFKA_HOST = settings.KAFKA_BROKER['HOST']
KAFKA_PORT = int(settings.KAFKA_BROKER['PORT'])
def get_device(self, serial): def get_device(self, serial):
try: try:
return Device.objects.get(serial=serial) return Device.objects.get(serial=serial)
@ -56,7 +56,9 @@ class Command(BaseCommand):
consumer = KafkaConsumer( consumer = KafkaConsumer(
"telemetry", "telemetry",
value_deserializer=lambda m: json.loads(m.decode('utf8')), value_deserializer=lambda m: json.loads(m.decode('utf8')),
bootstrap_servers='localhost:9092' bootstrap_servers='{}:{}'.format(
self.KAFKA_HOST, self.KAFKA_PORT
),
) )
break break
except NoBrokersAvailable: except NoBrokersAvailable: