I am using paho-mqtt in django to recieve messages. Everything works fine. But the on_message() function is executed twice.
I tried Debugging, but it seems like the function is called once, but the database insertion is happening twice, the printing of message is happening twice, everything within the on_message() function is happening twice, and my data is inserted twice for each publish.
I doubted it is happening in a parallel thread, and installed a celery redis backend to queue the insertion and avoid duplicate insertions. but still the data is being inserted twice.
I also tried locking the variables, to avoid problems in parallel threading, but still the data is inserted twice.
I am using Postgres DB
How do I solve this issue? I want the on_message() function to execute only once for each publish
my init.py
from . import mqtt
mqtt.client.loop_start()
my mqtt.py
import ast
import json
import paho.mqtt.client as mqtt
# Broker CONNACK response
from datetime import datetime
from raven.utils import logger
from kctsmarttransport import settings
def on_connect(client, userdata, flags, rc):
# Subcribing to topic and recoonect for
client.subscribe("data/gpsdata/server/#")
print 'subscribed to data/gpsdata/server/#'
# Receive message
def on_message(client, userdata, msg):
# from kctsmarttransport.celery import bus_position_insert_task
# bus_position_insert_task.delay(msg.payload)
from Transport.models import BusPosition
from Transport.models import Student, SpeedWarningLog, Bus
from Transport.models import Location
from Transport.models import IdleTimeLog
from pytz import timezone
try:
dumpData = json.dumps(msg.payload)
rawGpsData = json.loads(dumpData)
jsonGps = ast.literal_eval(rawGpsData)
bus = Bus.objects.get(bus_no=jsonGps['Busno'])
student = None
stop = None
if jsonGps['card'] is not False:
try:
student = Student.objects.get(rfid_value=jsonGps['UID'])
except Student.DoesNotExist:
student = None
if 'stop_id' in jsonGps:
stop = Location.objects.get(pk=jsonGps['stop_id'])
dates = datetime.strptime(jsonGps['Date&Time'], '%Y-%m-%d %H:%M:%S')
tz = timezone('Asia/Kolkata')
dates = tz.localize(dates)
lat = float(jsonGps['Latitude'])
lng = float(jsonGps['Longitude'])
speed = float(jsonGps['speed'])
# print msg.topic + " " + str(msg.payload)
busPosition = BusPosition.objects.filter(bus=bus, created_at=dates,
lat=lat,
lng=lng,
speed=speed,
geofence=stop,
student=student)
if busPosition.count() == 0:
busPosition = BusPosition.objects.create(bus=bus, created_at=dates,
lat=lat,
lng=lng,
speed=speed,
geofence=stop,
student=student)
if speed > 60:
SpeedWarningLog.objects.create(bus=busPosition.bus, speed=busPosition.speed,
lat=lat, lng=lng, created_at=dates)
sendSMS(settings.TRANSPORT_OFFICER_NUMBER, jsonGps['Busno'], jsonGps['speed'])
if speed <= 2:
try:
old_entry_query = IdleTimeLog.objects.filter(bus=bus, done=False).order_by('idle_start_time')
if old_entry_query.count() > 0:
old_entry = old_entry_query.reverse()[0]
old_entry.idle_end_time = dates
old_entry.save()
else:
new_entry = IdleTimeLog.objects.create(bus=bus, idle_start_time=dates, lat=lat, lng=lng)
except IdleTimeLog.DoesNotExist:
new_entry = IdleTimeLog.objects.create(bus=bus, idle_start_time=dates, lat=lat, lng=lng)
else:
try:
old_entry_query = IdleTimeLog.objects.filter(bus=bus, done=False).order_by('idle_start_time')
if old_entry_query.count() > 0:
old_entry = old_entry_query.reverse()[0]
old_entry.idle_end_time = dates
old_entry.done = True
old_entry.save()
except IdleTimeLog.DoesNotExist:
pass
except Exception, e:
logger.error(e.message, exc_info=True)
client = mqtt.Client()
client.on_connect = on_connect
client.on_message = on_message
client.connect("10.1.75.106", 1883, 60)
As some one mentioned in the comments run your server using --noreload
eg: python manage.py runserver --noreload
(posted here for better visibility.)