first

.env

+get_mongo=mongodb://intern_23:intern%40123@192.168.0.220:2717/?authSource=interns_b2_23&authMechanism=SCRAM-SHA-256
\ No newline at end of file

.idea/.gitignore

+# Default ignored files
+/shelf/
+/workspace.xml

.idea/.name

+main.py
\ No newline at end of file

.idea/inspectionProfiles/profiles_settings.xml

+<component name="InspectionProjectProfileManager">
+  <settings>
+    <option name="USE_PROJECT_PROFILE" value="false" />
+    <version value="1.0" />
+  </settings>
+</component>
\ No newline at end of file

.idea/misc.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectRootManager" version="2" project-jdk-name="C:\Users\ramya.r\Anaconda3" project-jdk-type="Python SDK" />
+</project>
\ No newline at end of file

.idea/modules.xml

+<?xml version="1.0" encoding="UTF-8"?>
+<project version="4">
+  <component name="ProjectModuleManager">
+    <modules>
+      <module fileurl="file://$PROJECT_DIR$/.idea/mqtt.iml" filepath="$PROJECT_DIR$/.idea/mqtt.iml" />
+    </modules>
+  </component>
+</project>
\ No newline at end of file

.idea/mqtt.iml

+<?xml version="1.0" encoding="UTF-8"?>
+<module type="PYTHON_MODULE" version="4">
+  <component name="NewModuleRootManager">
+    <content url="file://$MODULE_DIR$" />
+    <orderEntry type="inheritedJdk" />
+    <orderEntry type="sourceFolder" forTests="false" />
+  </component>
+</module>
\ No newline at end of file

conf/application.conf

+[MONGO_SERVICE]
+broker_host =192.168.0.220
+broker_port = 1883
+
+
+
+
+
+
+

main.py

+from script.core.engine.subscribe import MqttMongo
+
+get_file = MqttMongo()
+get_file.get_data()
\ No newline at end of file

publisher.py

+import json
+import time
+from datetime import datetime
+
+import paho.mqtt.client as mqtt
+
+from script.config.app_config import Service
+
+topic_one = "employee"
+message_one = {"emp_id": 101, "emp_name": "athira", "emp_department": "HR"}
+message_two = {"message send": datetime.utcnow().isoformat()}
+client = mqtt.Client()
+client.connect(Service.host, Service.port)
+while True:
+
+    if client.publish(topic_one, json.dumps(message_one)):
+        print(f"message published successfully{message_one}")
+        time.sleep(3)
+        client.disconnect()
+        break
+
+

script/config/app_config.py

+import os
+from dotenv import load_dotenv
+from configparser import SafeConfigParser
+
+
+config = SafeConfigParser()
+config.read('conf/application.conf')
+load_dotenv()
+
+
+class Service:
+    host = config.get("MONGO_SERVICE", "broker_host")
+    port = int(config.get("MONGO_SERVICE", "broker_port"))
+
+
+class Mongo:
+    mongo_uri: str = os.environ.get("get_mongo")
+

script/core/engine/subscribe.py

+
+from script.config.app_config import Service
+from script.core.interns.connection import connection_data
+import json
+import paho.mqtt.client as mqtt
+
+
+topic_one = "employee"
+class MqttMongo:
+    @staticmethod
+    def get_data():
+        def on_connect(client, userdata, flags, rc):
+            print("Connected with result code " + str(rc))
+            client.subscribe(topic_one)
+
+        def on_message(client_data, userdata, msg):
+            print("received" + msg.topic + " " + msg.payload.decode('utf-8'))
+            db = connection_data['interns_b2_23']
+            collection = db['ramya']
+            d = json.loads(msg.payload.decode('utf-8'))
+            temp = {'employee': msg.topic, 'payload': d}
+            t = collection.insert_one(temp)
+            print("saved" + str(t.inserted_id))
+
+        client = mqtt.Client()
+        client.on_connect = on_connect
+        client.on_message = on_message
+        client.connect(Service().host, Service().port)
+        client.loop_forever()

script/core/interns/connection.py

+import pymongo
+
+from script.config.app_config import Mongo
+
+connection_data = pymongo.MongoClient(Mongo.mongo_uri)
\ No newline at end of file