Django 通过 websocket 向所有客户端发送事件

Question

我想通过 websocket 向所有客户端（所有通过 websocket 建立连接的人）发送事件（json 字符串）。我的代码：

我觉得asgi和routing都可以

asgi.py

os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'base_app.settings')
django.setup()
application = ProtocolTypeRouter({
    "http": get_asgi_application(),
    "websocket": AuthMiddlewareStack(
        URLRouter(
            websocket_urlpatterns
        )
    ),
})

routing.py

from api_producer.consumers import EventWebSocket

websocket_urlpatterns = [
    url('ws/start/', EventWebSocket.as_asgi()),
]

问题点：一些函数生成事件，我想将这些事件发送给客户端

from api_producer.consumers import EventWebSocket

def event_producer():
    my_socket = EventWebSocket()
    my_socket.receive("Some_text")

consumers.py

from channels.generic.websocket import WebsocketConsumer
from asgiref.sync import async_to_sync


class EventWebSocket(WebsocketConsumer):
    def connect(self):
        self.channel_name = "CHANNEL_1"
        print("CALL_ACCEPT")
        async_to_sync(self.channel_layer.group_add)("global", self.channel_name)
        self.accept()

    def disconnect(self, close_code):
        async_to_sync(self.channel_layer.group_discard)("global", self.channel_name)

    def receive(self, text_data):
        print("RRR_ receive ", text_data)
        async_to_sync(self.channel_layer.group_send)(
            "global",
            {
                "type": "global.message",
                "text": text_data,
            },
        )

    def chat_message(self, event):
        print("CHAT_MESSAGE")
        self.send(text_data=event["text"])
    
    def global_event(self, event):
        print("GLOB_MESSAGE")
        self.send(text_data=event["text"])

我的行为： 1 到websocketKing 发送请求ws://127.0.0.1:8000/ws/start/。日志输出：

WebSocket HANDSHAKING /ws/start/ [127.0.0.1:37194]
WebSocket HANDSHAKING /ws/start/ [127.0.0.1:37194]
CALL_ACCEPT
HTTP POST /api/start_detection/ 200 [0.57, 127.0.0.1:37238] (trigger event_producer)

发送事件的正确方法是什么？

Answer 1

您应该通过通道层发送事件，you can use channel layers outside of consumers like this

from channels.layers import get_channel_layer
channel_layer = get_channel_layer()

def event_producer():
    async_to_sync(channel_layer. group_send)("global", {
        "type": "global.event",
        "text": "foo"
    })

并在您的消费者中加入事件生产者发送事件的全局通道，并添加一个可以由事件生产者触发的方法。您可以重复使用“聊天”频道...

class EventWebSocket(WebsocketConsumer):
    def connect(self):
        ...
        async_to_sync(self.channel_layer.group_add)("global", self.channel_name)
        self.accept()

    ...

    def global_event(self, event):
        self.send(text_data=event["text"])

Django 通过 websocket 向所有客户端发送事件

Django send events via websocket to all clients

python

django

websocket