Потребители¶

Закрытие потребителей¶

Когда сокет или соединение, подключенное к вашему потребителю, закрывается - вами или клиентом - вы, скорее всего, получите событие, отправленное вам (например, http.disconnect или << 1 >>>), и вашему экземпляру приложения будет предоставлено короткое время, чтобы действовать в соответствии с ним.

Как только вы закончите очистку после отключения, вам нужно поднять channels.exceptions.StopConsumer, чтобы остановить ASGI-приложение и позволить серверу очистить его. Если вы оставите его работать - не поднимите это исключение - сервер достигнет тайм-аута закрытия приложения (который в Daphne по умолчанию составляет 10 секунд), а затем убьет ваше приложение и выдаст предупреждение.

Приведенные ниже общие потребители делают это за вас, так что это необходимо только в том случае, если вы пишете свой собственный класс потребителя на основе AsyncConsumer или << 1 >>>. Однако, если вы переопределите их метод SyncConsumer или заблокируете методы обработки, которые он вызывает, от возврата, вы все равно можете столкнуться с этим; посмотрите их исходный код, если хотите получить больше информации.

Кроме того, если вы запускаете свои собственные фоновые программы, убедитесь, что вы также закрываете их по завершении соединения, иначе вы будете передавать программы на сервер.

Слои канала¶

Потребители также позволяют вам работать с канальными слоями Channel, чтобы позволить им отправлять сообщения между собой либо один на один, либо через широковещательную систему, называемую группами.

Потребители будут использовать канальный слой default, если не установлен атрибут channel_layer_alias при подклассификации любого из предоставленных классов Consumer.

Для использования слоя канала echo_alias мы установим его следующим образом:

from channels.consumer import SyncConsumer

class EchoConsumer(SyncConsumer):
    channel_layer_alias = "echo_alias"

Подробнее вы можете прочитать в Слои канала.

WebsocketConsumer¶

Доступно как channels.generic.websocket.WebsocketConsumer, это оборачивает многословную отправку и получение сообщений обычного ASGI в обработку, которая работает только с текстовыми и двоичными кадрами:

from channels.generic.websocket import WebsocketConsumer

class MyConsumer(WebsocketConsumer):
    groups = ["broadcast"]

    def connect(self):
        # Called on connection.
        # To accept the connection call:
        self.accept()
        # Or accept the connection and specify a chosen subprotocol.
        # A list of subprotocols specified by the connecting client
        # will be available in self.scope['subprotocols']
        self.accept("subprotocol")
        # To reject the connection, call:
        self.close()

    def receive(self, text_data=None, bytes_data=None):
        # Called with either text_data or bytes_data for each frame
        # You can call:
        self.send(text_data="Hello world!")
        # Or, to send a binary frame:
        self.send(bytes_data="Hello world!")
        # Want to force-close the connection? Call:
        self.close()
        # Or add a custom WebSocket error code!
        self.close(code=4123)

    def disconnect(self, close_code):
        # Called when the socket closes

Вы также можете поднять channels.exceptions.AcceptConnection или channels.exceptions.DenyConnection из любого места внутри метода connect, чтобы принять или отклонить соединение, если вам нужен многоразовый код аутентификации или ограничения скорости, не требующий использования микшинов.

Канал WebsocketConsumer будет автоматически добавлен (при подключении) и удален (при отключении) из всех групп, имена которых содержатся в атрибуте класса потребителя groups. groups должен быть итерабельным, а в качестве бэкенда канала должен быть установлен канальный уровень с поддержкой групп (channels.layers.InMemoryChannelLayer и channels_redis.core.RedisChannelLayer оба поддерживают группы). Если канальный уровень не настроен или канальный уровень не поддерживает группы, подключение к WebsocketConsumer с непустым атрибутом groups вызовет ошибку channels.exceptions.InvalidChannelLayerError. Подробнее см. в разделе Группы.

AsyncWebsocketConsumer¶

Доступный как channels.generic.websocket.AsyncWebsocketConsumer, он имеет точно такие же методы и сигнатуру, как WebsocketConsumer, но все асинхронно, и функции, которые вам нужно написать, также должны быть асинхронными:

from channels.generic.websocket import AsyncWebsocketConsumer

class MyConsumer(AsyncWebsocketConsumer):
    groups = ["broadcast"]

    async def connect(self):
        # Called on connection.
        # To accept the connection call:
        await self.accept()
        # Or accept the connection and specify a chosen subprotocol.
        # A list of subprotocols specified by the connecting client
        # will be available in self.scope['subprotocols']
        await self.accept("subprotocol")
        # To reject the connection, call:
        await self.close()

    async def receive(self, text_data=None, bytes_data=None):
        # Called with either text_data or bytes_data for each frame
        # You can call:
        await self.send(text_data="Hello world!")
        # Or, to send a binary frame:
        await self.send(bytes_data="Hello world!")
        # Want to force-close the connection? Call:
        await self.close()
        # Or add a custom WebSocket error code!
        await self.close(code=4123)

    async def disconnect(self, close_code):
        # Called when the socket closes

JsonWebsocketConsumer¶

Доступен как channels.generic.websocket.JsonWebsocketConsumer, работает как WebsocketConsumer, за исключением того, что он будет автоматически кодировать и декодировать в JSON, отправленный как текстовые фреймы WebSocket.

Различия только в API:

• Ваш метод receive_json должен принимать единственный аргумент content, то есть декодированный объект JSON.

• self.send_json принимает только один аргумент, content, который будет закодирован в JSON для вас.

Если вы хотите настроить кодирование и декодирование JSON, вы можете переопределить методы класса encode_json и << 1 >>>.

AsyncJsonWebsocketConsumer¶

Асинхронная версия JsonWebsocketConsumer, доступная как channels.generic.websocket.AsyncJsonWebsocketConsumer. Обратите внимание, что даже encode_json и decode_json являются асинхронными функциями.

AsyncHttpConsumer¶

Доступный как channels.generic.http.AsyncHttpConsumer, он предлагает базовые примитивы для реализации конечной точки HTTP:

from channels.generic.http import AsyncHttpConsumer

class BasicHttpConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        await asyncio.sleep(10)
        await self.send_response(200, b"Your response bytes", headers=[
            (b"Content-Type", b"text/plain"),
        ])

Ожидается, что вы реализуете свой собственный метод handle. Метод получает все тело запроса в виде одного байта. Заголовки могут быть переданы либо в виде списка кортежей, либо в виде словаря. Содержимое тела ответа ожидается как байтовая строка.

Вы также можете реализовать метод disconnect, если хотите запустить код при отключении - например, чтобы завершить все запущенные вами корутины. Этот метод будет выполняться даже при нечистом отключении, поэтому не ожидайте, что handle закончил выполнение чисто.

Если вам нужно больше контроля над ответом, например, для реализации длинного опроса, используйте вместо этого методы более низкого уровня self.send_headers и self.send_body. В этом примере уже упоминаются канальные слои, которые будут подробно описаны позже:

import json
from channels.generic.http import AsyncHttpConsumer

class LongPollConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        await self.send_headers(headers=[
            (b"Content-Type", b"application/json"),
        ])
        # Headers are only sent after the first body event.
        # Set "more_body" to tell the interface server to not
        # finish the response yet:
        await self.send_body(b"", more_body=True)

    async def chat_message(self, event):
        # Send JSON and finish the response:
        await self.send_body(json.dumps(event).encode("utf-8"))

Конечно, вы также можете использовать эти примитивы для реализации конечной точки HTTP для Server-sent events:

from datetime import datetime
from channels.generic.http import AsyncHttpConsumer

class ServerSentEventsConsumer(AsyncHttpConsumer):
    async def handle(self, body):
        await self.send_headers(headers=[
            (b"Cache-Control", b"no-cache"),
            (b"Content-Type", b"text/event-stream"),
            (b"Transfer-Encoding", b"chunked"),
        ])
        while True:
            payload = "data: %s\n\n" % datetime.now().isoformat()
            await self.send_body(payload.encode("utf-8"), more_body=True)
            await asyncio.sleep(1)