switch from WSGI using uWSGI to ASGI using custom event loop based on uvicorn

core/core/asgi.py

@@ -9,8 +9,25 @@ https://docs.djangoproject.com/en/4.2/howto/deployment/asgi/
 
 import os
 
+from channels.auth import AuthMiddlewareStack
+from channels.routing import ProtocolTypeRouter, URLRouter
+from channels.security.websocket import AllowedHostsOriginValidator
 from django.core.asgi import get_asgi_application
+from notify_sessions.routing import websocket_urlpatterns
 
 os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'core.settings')
 
-application = get_asgi_application()
+django_asgi_app = get_asgi_application()
+
+websocket_asgi_app = AllowedHostsOriginValidator(
+    #AuthMiddlewareStack(
+        URLRouter(
+            websocket_urlpatterns
+        )
+    #)
+)
+
+application = ProtocolTypeRouter({
+    "http": django_asgi_app,
+    "websocket": websocket_asgi_app,
+})

core/core/settings.py

@@ -49,6 +49,7 @@ INSTALLED_APPS = [
     'channels',
     'files',
     'inventory',
+    'notify_sessions',
 ]
 
 REST_FRAMEWORK = {
@@ -183,12 +184,10 @@ DEFAULT_AUTO_FIELD = 'django.db.models.BigAutoField'
 
 CHANNEL_LAYERS = {
     'default': {
-        'BACKEND': 'channels.layers.InMemoryChannelLayer',
-        # 'BACKEND': 'asgi_redis.RedisChannelLayer',
-        # 'CONFIG': {
-        #    'hosts': [('localhost', 6379)],
-        # },
-        'ROUTING': 'example.routing.channel_routing',
+        'BACKEND': 'asgi_redis.RedisChannelLayer',
+        'CONFIG': {
+            'hosts': [('localhost', 6379)],
+        },
     }
 
 }

core/notify_sessions/consumers.py

@@ -0,0 +1,40 @@
+import json
+
+from channels.generic.websocket import AsyncWebsocketConsumer
+
+
+class NotifyConsumer(AsyncWebsocketConsumer):
+    def __init__(self, *args, **kwargs):
+        super().__init__(args, kwargs)
+        self.room_group_name = None
+        self.event_slug = None
+
+    async def connect(self):
+        self.event_slug = self.scope["url_route"]["kwargs"]["event_slug"]
+        self.room_group_name = f"chat_{self.event_slug}"
+
+        # Join room group
+        await self.channel_layer.group_add(self.room_group_name, self.channel_name)
+
+        await self.accept()
+
+    async def disconnect(self, close_code):
+        # Leave room group
+        await self.channel_layer.group_discard(self.room_group_name, self.channel_name)
+
+    # Receive message from WebSocket
+    async def receive(self, text_data):
+        text_data_json = json.loads(text_data)
+        message = text_data_json["message"]
+
+        # Send message to room group
+        await self.channel_layer.group_send(
+            self.room_group_name, {"type": "chat.message", "message": message}
+        )
+
+    # Receive message from room group
+    async def chat_message(self, event):
+        message = event["message"]
+
+        # Send message to WebSocket
+        await self.send(text_data=json.dumps({"message": message}))

core/notify_sessions/routing.py

@@ -0,0 +1,8 @@
+from django.urls import path
+
+from .consumers import NotifyConsumer
+
+websocket_urlpatterns = [
+    path('notify/<str:event_slug>/', NotifyConsumer.as_asgi()),
+]
+

core/requirements.txt

@@ -18,7 +18,6 @@ itypes==1.2.0
 Jinja2==3.1.2
 MarkupSafe==2.1.3
 msgpack-python==0.5.6
-mysqlclient
 openapi-codec==1.3.2
 packaging==23.2
 Pillow==10.1.0
@@ -27,9 +26,13 @@ pytz==2023.3.post1
 PyYAML==6.0.1
 redis
 requests==2.31.0
+sdnotify==0.3.2
+setproctitle==1.3.3
 six==1.16.0
 sqlparse==0.4.4
 typing_extensions==4.8.0
 uritemplate==4.1.1
 urllib3==2.1.0
+uvicorn==0.24.0.post1
+watchfiles==0.21.0
 websockets==12.0

core/server.py

@@ -0,0 +1,130 @@
+#!/usr/bin/env python3
+import asyncio
+import logging
+
+import uvicorn
+from aiosmtpd.controller import Controller, UnixSocketController
+from aiosmtpd.lmtp import LMTP
+
+
+async def handle_echo(reader, writer):
+    # session = Session()
+    data = await reader.read(100)
+    message = data.decode()
+    addr = writer.get_extra_info('peername')
+
+    print(f"Received {message!r} from {addr!r}")
+
+    print(f"Send: {message!r}")
+    writer.write(data)
+    await writer.drain()
+
+    print("Close the connection")
+    writer.close()
+    await writer.wait_closed()
+
+
+class ExampleHandler:
+    async def handle_RCPT(self, server, session, envelope, address, rcpt_options):
+        if not address.endswith('@example.com'):
+            return '550 not relaying to that domain'
+        envelope.rcpt_tos.append(address)
+        return '250 OK'
+
+    async def handle_DATA(self, server, session, envelope):
+        print('Message from %s' % envelope.mail_from)
+        print('Message for %s' % envelope.rcpt_tos)
+        print('Message data:\n')
+        for ln in envelope.content.decode('utf8', errors='replace').splitlines():
+            print(f'> {ln}'.strip())
+        print()
+        print('End of message')
+        return '250 Message accepted for delivery'
+
+
+class LTMPController(Controller):
+    def factory(self):
+        return LMTP(self.handler)
+
+
+class UnixSocketLMTPController(UnixSocketController):
+    def factory(self):
+        return LMTP(self.handler)
+
+
+class UvicornServer(uvicorn.Server):
+    def install_signal_handlers(self):
+        pass
+
+
+async def web():
+    log_config = uvicorn.config.LOGGING_CONFIG
+    log_config["handlers"]["default"] = {"class": "logging.FileHandler", "filename": "web.log", "formatter": "default"}
+    log_config["handlers"]["access"] = {"class": "logging.FileHandler", "filename": "web-access.log",
+                                        "formatter": "access"}
+    config = uvicorn.Config("core.asgi:application", uds="web.sock", log_config=log_config)
+    server = UvicornServer(config=config)
+    await server.serve()
+
+
+async def tcp():
+    log = logging.getLogger('test')
+    log.info("Starting TCP server")
+    server = await asyncio.start_unix_server(handle_echo, path='test.sock')
+
+    addrs = ', '.join(str(sock.getsockname()) for sock in server.sockets)
+    log.info(f'Serving on {addrs}')
+
+    async with server:
+        await server.serve_forever()
+    log.info("TCP done")
+
+
+async def lmtp():
+    log = logging.getLogger('lmtp')
+    log.info("Starting LMTP server")
+    cont = UnixSocketLMTPController(ExampleHandler(), unix_socket='lmtp.sock')
+    cont.start()
+    log.info("LMTP done")
+
+
+async def shutdown(sig, loop):
+    log = logging.getLogger()
+    log.info(f"Received exit signal {sig.name}...")
+    tasks = [t for t in asyncio.all_tasks() if t is not
+             asyncio.current_task()]
+    [task.cancel() for task in tasks]
+    log.info(f"Cancelling {len(tasks)} outstanding tasks")
+    await asyncio.wait_for(loop.shutdown_asyncgens(), timeout=10)
+    loop.stop()
+    log.info("Shutdown complete.")
+
+
+def main():
+    import sdnotify
+    import signal
+    import setproctitle
+    setproctitle.setproctitle("c3lf-sys3")
+    logging.basicConfig(filename='server.log', level=logging.DEBUG, encoding='utf-8')
+    logging.basicConfig(filename='test.log', level=logging.DEBUG, encoding='utf-8')
+    logging.basicConfig(filename='lmtp.log', level=logging.DEBUG, encoding='utf-8')
+    log = logging.getLogger()
+    log.info("Starting server")
+    loop = asyncio.get_event_loop()
+    loop.add_signal_handler(signal.SIGTERM, lambda: asyncio.create_task(shutdown(signal.SIGTERM, loop)))
+    loop.add_signal_handler(signal.SIGINT, lambda: asyncio.create_task(shutdown(signal.SIGINT, loop)))
+    loop.create_task(web())
+    loop.create_task(tcp())
+    loop.create_task(lmtp())
+    n = sdnotify.SystemdNotifier()
+    n.notify("READY=1")
+    log.info("Server ready")
+    try:
+        loop.run_forever()
+    finally:
+        loop.close()
+    logging.info("Server stopped")
+
+
+if __name__ == '__main__':
+    main()