Marmelator/pyinfra
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases 36 Wiki Activity

feat: wip for multiple tenants - for pkg build

Browse Source
This commit is contained in:
Jonathan Kössler 2024-07-10 11:45:47 +02:00
parent b2e3ae092f
commit 6e7c4ccb7b
2 changed files with 13 additions and 13 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

24
pyinfra/examples.py
View File

@ -37,8 +37,8 @@ def start_standard_queue_consumer(
app = app or FastAPI()
# tenant_manager = TenantQueueManager(settings)
# service_manager = ServiceQueueManager(settings)
tenant_manager = TenantQueueManager(settings)
service_manager = ServiceQueueManager(settings)
if settings.metrics.prometheus.enabled:
logger.info("Prometheus metrics enabled.")
@ -51,21 +51,21 @@ def start_standard_queue_consumer(
instrument_pika()
instrument_app(app)
manager = AsyncQueueManager(settings=settings, message_processor=callback)
# manager = AsyncQueueManager(settings=settings, message_processor=callback)
# app = add_health_check_endpoint(app, service_manager.is_ready)
app = add_health_check_endpoint(app, manager.is_ready)
app = add_health_check_endpoint(app, service_manager.is_ready)
# app = add_health_check_endpoint(app, manager.is_ready)
webserver_thread = create_webserver_thread_from_settings(app, settings)
webserver_thread.start()
# thread_t = Thread(target=tenant_manager.start_consuming, daemon=True)
# thread_s = Thread(target=service_manager.start_sequential_basic_get, args=(callback,), daemon=True)
thread_t = Thread(target=tenant_manager.start_consuming, daemon=True)
thread_s = Thread(target=service_manager.start_sequential_basic_get, args=(callback,), daemon=True)
# thread_t.start()
# thread_s.start()
thread_t.start()
thread_s.start()
# thread_t.join()
# thread_s.join()
thread_t.join()
thread_s.join()
asyncio.run(manager.start_processing())
# asyncio.run(manager.start_processing())

2
pyinfra/queue/threaded_tenants.py
View File

@ -312,7 +312,7 @@ class ServiceQueueManager(BaseQueueManager):
on_message_callback = self._make_on_message_callback(message_processor, tenant_id)
on_message_callback(self.channel, method_frame, properties, body)
else:
logger.debug(f"No message returned for queue {queue_name}")
# logger.debug(f"No message returned for queue {queue_name}")
# time.sleep(self.connection_sleep)
time.sleep(0.1)