rabbitmq tests

main.py

@@ -1,39 +1,103 @@
-import datetime
-from time import sleep, time
-
-from multiprocessing import Process, Queue
-import source.vqe as vqe
-#import project.source.archive.mqtt as mqtt
-
-if __name__ == "__main__":
-    
-
-    broker_address = "mqtt.deowl.ru" # Example public broker
-    broker_port = 1883
-    #mqttBroker = mqtt.vqeMqttBroker(broker_address, broker_port, max_connection_attempts=5)
-    #print(mqttBroker.get_status())
-    #mqttBroker.connect_to_server()
-
-    #while True:
-    #    print(mqttBroker.get_status())
-    #    sleep(1)
-
-    # GLOBAL SYSTEM PARAMETERS
-    #active_electrons=2
-    '''active_orbitals=2
-    max_iterations = 500
-    conv_tol = 1e-04
-    step_size = 0.05
-
-    q = Queue()
-    p = Process(target=vqe.run_vqe, args=(q, symbols, coordinates, active_electrons, active_orbitals, max_iterations, conv_tol, step_size))
-    p.start()
-    while p.is_alive():
-        try:
-            print(q.get_nowait(), datetime.datetime.now())
-        except Exception as e:
-            print("no_data_to_get", e)
-        sleep(0.1)
-    
-    client.loop_stop() # Stop the background loop thread if used
-    client.disconnect()'''
+import asyncio
+import datetime
+import logging
+import os
+from ast import Lambda
+from multiprocessing import Process, Queue
+from socket import timeout
+from time import sleep, time
+
+import aio_pika
+import source.vqe as vqe
+from fasthtml.common import Div, P, fast_app, serve
+
+logger = logging.basicConfig(level=logging.INFO)
+app, rt = fast_app()
+
+
+async def rabbit_worker():
+    broker_address = os.environ.get("MQTT_HOST")  # Example public broker
+    broker_port = os.environ.get("MQTT_PORT")
+
+    if not broker_address or not broker_port:
+        if not broker_address:
+            logging.fatal("Not Found Environment Variable: MQTT_HOST")
+        if not broker_port:
+            logging.fatal("Not Found Environment Variable: MQTT_PORT")
+        logging.info("Shutting down")
+        return
+    logging.info(f"starting connection: {broker_address}, {broker_port}")
+    connection = await aio_pika.connect(
+        host=broker_address, port=int(broker_port), timeout=5
+    )
+    logging.info("debug")
+    channel = await connection.channel()
+    queue = await channel.declare_queue("my_queue", durable=True)
+
+    async with queue.iterator() as queue_iter:
+        async for message in queue_iter:
+            async with message.process():
+                asyncio.create_task(handle_message(message.body))
+
+
+async def handle_message(body):
+    print("Processing:", body)
+    await asyncio.sleep(2)  # simulate work
+
+
+def handle_done(body):
+    logging.info(f"done, {body}")
+
+
+@app.on_event("startup")
+async def startup():
+    global rabbit_task
+    rabbit_task = asyncio.create_task(rabbit_worker())
+    rabbit_task.add_done_callback(handle_done)
+
+
+@app.on_event("shutdown")
+async def shutdown():
+    rabbit_task.cancel()
+
+
+@rt("/")
+def get():
+    return Div(P("Hello World!"), hx_get="/change")
+
+
+if __name__ == "__main__":
+    serve()
+
+    """
+    # GLOBAL SYSTEM PARAMETERS
+    active_electrons = 2
+    active_orbitals = 2
+    max_iterations = 500
+    conv_tol = 1e-04
+    step_size = 0.05
+
+    q = Queue()
+    p = Process(
+        target=vqe.run_vqe,
+        args=(
+            q,
+            symbols,
+            coordinates,
+            active_electrons,
+            active_orbitals,
+            max_iterations,
+            conv_tol,
+            step_size,
+        ),
+    )
+    p.start()
+    while p.is_alive():
+        try:
+            print(q.get_nowait(), datetime.datetime.now())
+        except Exception as e:
+            print("no_data_to_get", e)
+        sleep(0.1)
+
+    # client.loop_stop() # Stop the background loop thread if used
+    # client.disconnect()"""