From db0d2975a0a92082dd7b0f7b11ec5207f5e74c93 Mon Sep 17 00:00:00 2001 From: Andrew Godwin Date: Wed, 5 Oct 2016 12:06:34 -0700 Subject: [PATCH] Remove middleware approach, change to simpler one --- channels/channel.py | 2 - channels/consumer_middleware.py | 91 --------------------------------- channels/signals.py | 4 -- channels/worker.py | 12 +++-- docs/asgi.rst | 3 ++ 5 files changed, 10 insertions(+), 102 deletions(-) delete mode 100644 channels/consumer_middleware.py diff --git a/channels/channel.py b/channels/channel.py index a308e9e..b65d65a 100644 --- a/channels/channel.py +++ b/channels/channel.py @@ -3,7 +3,6 @@ from __future__ import unicode_literals from django.utils import six from channels import DEFAULT_CHANNEL_LAYER, channel_layers -from .signals import message_sent class Channel(object): @@ -37,7 +36,6 @@ class Channel(object): if not isinstance(content, dict): raise TypeError("You can only send dicts as content on channels.") self.channel_layer.send(self.name, content) - message_sent.send(sender=self.__class__, channel=self.name, keys=list(content.keys())) def __str__(self): return self.name diff --git a/channels/consumer_middleware.py b/channels/consumer_middleware.py deleted file mode 100644 index 34284dc..0000000 --- a/channels/consumer_middleware.py +++ /dev/null @@ -1,91 +0,0 @@ -from __future__ import unicode_literals - -import importlib -import threading -import warnings -from django.conf import settings - -from .exceptions import DenyConnection -from .signals import consumer_started, consumer_finished, message_sent - - -class ConsumerMiddlewareRegistry(object): - """ - Handles registration (via settings object) and generation of consumer - middleware stacks - """ - - fixed_middleware = ["channels.consumer_middleware.ConvenienceMiddleware"] - - def __init__(self): - # Load middleware callables from settings - middleware_paths = self.fixed_middleware + getattr(settings, "CONSUMER_MIDDLEWARE", []) - self.middleware_instances = [] - for path in middleware_paths: - module_name, variable_name = path.rsplit(".", 1) - try: - self.middleware_instances.append(getattr(importlib.import_module(module_name), variable_name)) - except (ImportError, AttributeError) as e: - raise ImproperlyConfigured("Cannot import consumer middleware %r: %s" % (path, e)) - - def make_chain(self, consumer, kwargs): - """ - Returns an instantiated chain of middleware around a final consumer. - """ - next_layer = lambda message: consumer(message, **kwargs) - for middleware_instance in reversed(self.middleware_instances): - next_layer = middleware_instance(next_layer) - return next_layer - - -class ConvenienceMiddleware(object): - """ - Standard middleware which papers over some more explicit parts of ASGI. - """ - - runtime_data = threading.local() - - def __init__(self, consumer): - self.consumer = consumer - - def __call__(self, message): - if message.channel.name == "websocket.connect": - # Websocket connect acceptance helper - try: - self.consumer(message) - except DenyConnection: - message.reply_channel.send({"accept": False}) - else: - replies_sent = [msg for chan, msg in self.get_messages() if chan == message.reply_channel.name] - # If they sent no replies, send implicit acceptance - if not replies_sent: - warnings.warn("AAAAAAAAAAA", RuntimeWarning) - message.reply_channel.send({"accept": True}) - else: - # General path - return self.consumer(message) - - @classmethod - def reset_messages(cls, **kwargs): - """ - Tied to the consumer started/ended signal to reset the messages list. - """ - cls.runtime_data.sent_messages = [] - - consumer_started.connect(lambda **kwargs: ConvenienceMiddleware.reset_messages(), weak=False) - consumer_finished.connect(lambda **kwargs: ConvenienceMiddleware.reset_messages(), weak=False) - - @classmethod - def sent_message(cls, channel, keys, **kwargs): - """ - Called by message sending interfaces when messages are sent, - for convenience errors only. Should not be relied upon to get - all messages. - """ - cls.runtime_data.sent_messages = getattr(cls.runtime_data, "sent_messages", []) + [(channel, keys)] - - message_sent.connect(lambda channel, keys, **kwargs: ConvenienceMiddleware.sent_message(channel, keys), weak=False) - - @classmethod - def get_messages(cls): - return getattr(cls.runtime_data, "sent_messages", []) diff --git a/channels/signals.py b/channels/signals.py index 0a0e575..dc83b94 100644 --- a/channels/signals.py +++ b/channels/signals.py @@ -7,9 +7,5 @@ consumer_finished = Signal() worker_ready = Signal() worker_process_ready = Signal() -# Called when a message is sent directly to a channel. Not called for group -# sends or direct ASGI usage. For convenience/nicer errors only. -message_sent = Signal(providing_args=["channel", "keys"]) - # Connect connection closer to consumer finished as well consumer_finished.connect(close_old_connections) diff --git a/channels/worker.py b/channels/worker.py index 4f48615..bad22df 100644 --- a/channels/worker.py +++ b/channels/worker.py @@ -9,11 +9,10 @@ import multiprocessing import threading from .signals import consumer_started, consumer_finished -from .exceptions import ConsumeLater +from .exceptions import ConsumeLater, DenyConnection from .message import Message from .utils import name_that_thing from .signals import worker_ready -from .consumer_middleware import ConsumerMiddlewareRegistry logger = logging.getLogger('django.channels') @@ -41,7 +40,6 @@ class Worker(object): self.exclude_channels = exclude_channels self.termed = False self.in_job = False - self.middleware_registry = ConsumerMiddlewareRegistry() def install_signal_handler(self): signal.signal(signal.SIGTERM, self.sigterm_handler) @@ -119,8 +117,12 @@ class Worker(object): # Send consumer started to manage lifecycle stuff consumer_started.send(sender=self.__class__, environ={}) # Run consumer - chain = self.middleware_registry.make_chain(consumer, kwargs) - chain(message) + consumer(message, **kwargs) + except DenyConnection: + # They want to deny a WebSocket connection. + if message.channel.name != "websocket.connect": + raise ValueError("You cannot DenyConnection from a non-websocket.connect handler.") + message.reply_channel.send({"accept": False}) except ConsumeLater: # They want to not handle it yet. Re-inject it with a number-of-tries marker. content['__retries__'] = content.get("__retries__", 0) + 1 diff --git a/docs/asgi.rst b/docs/asgi.rst index 919d077..0a5b68b 100644 --- a/docs/asgi.rst +++ b/docs/asgi.rst @@ -790,6 +790,9 @@ is received to say if the connection should be accepted or dropped. Behaviour on WebSocket rejection is defined in the Connection section above. +If received while the socket is already accepted, the protocol server should +log an error, but not do anything. + Channel: ``websocket.send!`` Keys: