django/daphne
1
1
Fork 0
mirror of https://github.com/django/daphne.git synced 2025-11-04 09:37:32 +03:00
Code Issues Packages Projects Releases Wiki Activity

Fix line endings

Browse Source
This commit is contained in:
Andrew Godwin 2015-06-10 09:42:34 -07:00
parent c9eb683ed8
commit 95e706f71f
12 changed files with 453 additions and 453 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

36
channels/__init__.py
View File

@ -1,18 +1,18 @@
# Load backends # Load backends
DEFAULT_CHANNEL_BACKEND = "default" DEFAULT_CHANNEL_BACKEND = "default"
from .backends import BackendManager from .backends import BackendManager
from django.conf import settings from django.conf import settings
channel_backends = BackendManager( channel_backends = BackendManager(
getattr(settings, "CHANNEL_BACKENDS", { getattr(settings, "CHANNEL_BACKENDS", {
DEFAULT_CHANNEL_BACKEND: { DEFAULT_CHANNEL_BACKEND: {
"BACKEND": "channels.backends.memory.InMemoryChannelBackend", "BACKEND": "channels.backends.memory.InMemoryChannelBackend",
} }
}) })
) )
# Ensure monkeypatching # Ensure monkeypatching
from .hacks import monkeypatch_django from .hacks import monkeypatch_django
monkeypatch_django() monkeypatch_django()
# Promote channel to top-level (down here to avoid circular import errs) # Promote channel to top-level (down here to avoid circular import errs)
from .channel import Channel from .channel import Channel

106
channels/adapters.py
View File

@ -1,53 +1,53 @@
import functools import functools
from django.core.handlers.base import BaseHandler from django.core.handlers.base import BaseHandler
from django.http import HttpRequest, HttpResponse from django.http import HttpRequest, HttpResponse
from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND
class UrlConsumer(object): class UrlConsumer(object):
""" """
Dispatches channel HTTP requests into django's URL system. Dispatches channel HTTP requests into django's URL system.
""" """
def __init__(self): def __init__(self):
self.handler = BaseHandler() self.handler = BaseHandler()
self.handler.load_middleware() self.handler.load_middleware()
def __call__(self, channel, **kwargs): def __call__(self, channel, **kwargs):
request = HttpRequest.channel_decode(kwargs) request = HttpRequest.channel_decode(kwargs)
try: try:
response = self.handler.get_response(request) response = self.handler.get_response(request)
except HttpResponse.ResponseLater: except HttpResponse.ResponseLater:
return return
Channel(request.response_channel).send(**response.channel_encode()) Channel(request.response_channel).send(**response.channel_encode())
def view_producer(channel_name): def view_producer(channel_name):
""" """
Returns a new view function that actually writes the request to a channel Returns a new view function that actually writes the request to a channel
and abandons the response (with an exception the Worker will catch) and abandons the response (with an exception the Worker will catch)
""" """
def producing_view(request): def producing_view(request):
Channel(channel_name).send(**request.channel_encode()) Channel(channel_name).send(**request.channel_encode())
raise HttpResponse.ResponseLater() raise HttpResponse.ResponseLater()
return producing_view return producing_view
def view_consumer(channel_name, alias=DEFAULT_CHANNEL_BACKEND): def view_consumer(channel_name, alias=DEFAULT_CHANNEL_BACKEND):
""" """
Decorates a normal Django view to be a channel consumer. Decorates a normal Django view to be a channel consumer.
Does not run any middleware Does not run any middleware
""" """
def inner(func): def inner(func):
@functools.wraps(func) @functools.wraps(func)
def consumer(channel, **kwargs): def consumer(channel, **kwargs):
request = HttpRequest.channel_decode(kwargs) request = HttpRequest.channel_decode(kwargs)
response = func(request) response = func(request)
Channel(request.response_channel).send(**response.channel_encode()) Channel(request.response_channel).send(**response.channel_encode())
# Get the channel layer and register # Get the channel layer and register
channel_layer = channel_backends[DEFAULT_CHANNEL_BACKEND] channel_layer = channel_backends[DEFAULT_CHANNEL_BACKEND]
channel_layer.registry.add_consumer(consumer, [channel_name]) channel_layer.registry.add_consumer(consumer, [channel_name])
return func return func
return inner return inner

66
channels/backends/base.py
View File

@ -1,33 +1,33 @@
from channels.consumer_registry import ConsumerRegistry from channels.consumer_registry import ConsumerRegistry
class ChannelClosed(Exception): class ChannelClosed(Exception):
""" """
Raised when you try to send to a closed channel. Raised when you try to send to a closed channel.
""" """
pass pass
class BaseChannelBackend(object): class BaseChannelBackend(object):
""" """
Base class for all channel layer implementations. Manages both sending Base class for all channel layer implementations. Manages both sending
and receving messages from the backend, and each comes with its own and receving messages from the backend, and each comes with its own
registry of consumers. registry of consumers.
""" """
def __init__(self, expiry=60): def __init__(self, expiry=60):
self.registry = ConsumerRegistry() self.registry = ConsumerRegistry()
self.expiry = expiry self.expiry = expiry
def send(self, channel, message): def send(self, channel, message):
""" """
Send a message over the channel, taken from the kwargs. Send a message over the channel, taken from the kwargs.
""" """
raise NotImplementedError() raise NotImplementedError()
def receive_many(self, channels): def receive_many(self, channels):
""" """
Block and return the first message available on one of the Block and return the first message available on one of the
channels passed, as a (channel, message) tuple. channels passed, as a (channel, message) tuple.
""" """
raise NotImplementedError() raise NotImplementedError()

64
channels/backends/memory.py
View File

@ -1,32 +1,32 @@
import time import time
import json import json
from collections import deque from collections import deque
from .base import BaseChannelBackend from .base import BaseChannelBackend
queues = {} queues = {}
class InMemoryChannelBackend(BaseChannelBackend): class InMemoryChannelBackend(BaseChannelBackend):
""" """
In-memory channel implementation. Intended only for use with threading, In-memory channel implementation. Intended only for use with threading,
in low-throughput development environments. in low-throughput development environments.
""" """
def send(self, channel, message): def send(self, channel, message):
# Try JSON encoding it to make sure it would, but store the native version # Try JSON encoding it to make sure it would, but store the native version
json.dumps(message) json.dumps(message)
# Add to the deque, making it if needs be # Add to the deque, making it if needs be
queues.setdefault(channel, deque()).append(message) queues.setdefault(channel, deque()).append(message)
def receive_many(self, channels): def receive_many(self, channels):
while True: while True:
# Try to pop a message from each channel # Try to pop a message from each channel
for channel in channels: for channel in channels:
try: try:
# This doesn't clean up empty channels - OK for testing. # This doesn't clean up empty channels - OK for testing.
# For later versions, have cleanup w/lock. # For later versions, have cleanup w/lock.
return channel, queues[channel].popleft() return channel, queues[channel].popleft()
except (IndexError, KeyError): except (IndexError, KeyError):
pass pass
# If all empty, sleep for a little bit # If all empty, sleep for a little bit
time.sleep(0.01) time.sleep(0.01)

136
channels/backends/orm.py
View File

@ -1,68 +1,68 @@
import time import time
import datetime import datetime
from django.apps.registry import Apps from django.apps.registry import Apps
from django.db import models, connections, DEFAULT_DB_ALIAS from django.db import models, connections, DEFAULT_DB_ALIAS
from .base import BaseChannelBackend from .base import BaseChannelBackend
queues = {} queues = {}
class ORMChannelBackend(BaseChannelBackend): class ORMChannelBackend(BaseChannelBackend):
""" """
ORM-backed channel environment. For development use only; it will span ORM-backed channel environment. For development use only; it will span
multiple processes fine, but it's going to be pretty bad at throughput. multiple processes fine, but it's going to be pretty bad at throughput.
""" """
def __init__(self, expiry, db_alias=DEFAULT_DB_ALIAS): def __init__(self, expiry, db_alias=DEFAULT_DB_ALIAS):
super(ORMChannelBackend, self).__init__(expiry) super(ORMChannelBackend, self).__init__(expiry)
self.connection = connections[db_alias] self.connection = connections[db_alias]
self.model = self.make_model() self.model = self.make_model()
self.ensure_schema() self.ensure_schema()
def make_model(self): def make_model(self):
""" """
Initialises a new model to store messages; not done as part of a Initialises a new model to store messages; not done as part of a
models.py as we don't want to make it for most installs. models.py as we don't want to make it for most installs.
""" """
class Message(models.Model): class Message(models.Model):
# We assume an autoincrementing PK for message order # We assume an autoincrementing PK for message order
channel = models.CharField(max_length=200, db_index=True) channel = models.CharField(max_length=200, db_index=True)
content = models.TextField() content = models.TextField()
expiry = models.DateTimeField(db_index=True) expiry = models.DateTimeField(db_index=True)
class Meta: class Meta:
apps = Apps() apps = Apps()
app_label = "channels" app_label = "channels"
db_table = "django_channels" db_table = "django_channels"
return Message return Message
def ensure_schema(self): def ensure_schema(self):
""" """
Ensures the table exists and has the correct schema. Ensures the table exists and has the correct schema.
""" """
# If the table's there, that's fine - we've never changed its schema # If the table's there, that's fine - we've never changed its schema
# in the codebase. # in the codebase.
if self.model._meta.db_table in self.connection.introspection.table_names(self.connection.cursor()): if self.model._meta.db_table in self.connection.introspection.table_names(self.connection.cursor()):
return return
# Make the table # Make the table
with self.connection.schema_editor() as editor: with self.connection.schema_editor() as editor:
editor.create_model(self.model) editor.create_model(self.model)
def send(self, channel, message): def send(self, channel, message):
self.model.objects.create( self.model.objects.create(
channel = channel, channel = channel,
message = json.dumps(message), message = json.dumps(message),
expiry = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.expiry) expiry = datetime.datetime.utcnow() + datetime.timedelta(seconds=self.expiry)
) )
def receive_many(self, channels): def receive_many(self, channels):
while True: while True:
# Delete all expired messages (add 10 second grace period for clock sync) # Delete all expired messages (add 10 second grace period for clock sync)
self.model.objects.filter(expiry__lt=datetime.datetime.utcnow() - datetime.timedelta(seconds=10)).delete() self.model.objects.filter(expiry__lt=datetime.datetime.utcnow() - datetime.timedelta(seconds=10)).delete()
# Get a message from one of our channels # Get a message from one of our channels
message = self.model.objects.filter(channel__in=channels).order_by("id").first() message = self.model.objects.filter(channel__in=channels).order_by("id").first()
if message: if message:
return message.channel, json.loads(message.content) return message.channel, json.loads(message.content)
# If all empty, sleep for a little bit # If all empty, sleep for a little bit
time.sleep(0.2) time.sleep(0.2)

80
channels/consumer_registry.py
View File

@ -1,40 +1,40 @@
import functools import functools
from django.utils import six from django.utils import six
from .utils import name_that_thing from .utils import name_that_thing
class ConsumerRegistry(object): class ConsumerRegistry(object):
""" """
Manages the available consumers in the project and which channels they Manages the available consumers in the project and which channels they
listen to. listen to.
Generally this is attached to a backend instance as ".registry" Generally this is attached to a backend instance as ".registry"
""" """
def __init__(self): def __init__(self):
self.consumers = {} self.consumers = {}
def add_consumer(self, consumer, channels): def add_consumer(self, consumer, channels):
# Upconvert if you just pass in a string # Upconvert if you just pass in a string
if isinstance(channels, six.string_types): if isinstance(channels, six.string_types):
channels = [channels] channels = [channels]
# Register on each channel, checking it's unique # Register on each channel, checking it's unique
for channel in channels: for channel in channels:
if channel in self.consumers: if channel in self.consumers:
raise ValueError("Cannot register consumer %s - channel %r already consumed by %s" % ( raise ValueError("Cannot register consumer %s - channel %r already consumed by %s" % (
name_that_thing(consumer), name_that_thing(consumer),
channel, channel,
name_that_thing(self.consumers[channel]), name_that_thing(self.consumers[channel]),
)) ))
self.consumers[channel] = consumer self.consumers[channel] = consumer
def all_channel_names(self): def all_channel_names(self):
return self.consumers.keys() return self.consumers.keys()
def consumer_for_channel(self, channel): def consumer_for_channel(self, channel):
try: try:
return self.consumers[channel] return self.consumers[channel]
except KeyError: except KeyError:
return None return None

54
channels/hacks.py
View File

@ -1,27 +1,27 @@
from django.http.request import HttpRequest from django.http.request import HttpRequest
from django.http.response import HttpResponseBase from django.http.response import HttpResponseBase
from django.core.handlers.base import BaseHandler from django.core.handlers.base import BaseHandler
from .request import encode_request, decode_request from .request import encode_request, decode_request
from .response import encode_response, decode_response, ResponseLater from .response import encode_response, decode_response, ResponseLater
def monkeypatch_django(): def monkeypatch_django():
""" """
Monkeypatches support for us into parts of Django. Monkeypatches support for us into parts of Django.
""" """
# Request encode/decode # Request encode/decode
HttpRequest.channel_encode = encode_request HttpRequest.channel_encode = encode_request
HttpRequest.channel_decode = staticmethod(decode_request) HttpRequest.channel_decode = staticmethod(decode_request)
# Response encode/decode # Response encode/decode
HttpResponseBase.channel_encode = encode_response HttpResponseBase.channel_encode = encode_response
HttpResponseBase.channel_decode = staticmethod(decode_response) HttpResponseBase.channel_decode = staticmethod(decode_response)
HttpResponseBase.ResponseLater = ResponseLater HttpResponseBase.ResponseLater = ResponseLater
# Allow ResponseLater to propagate above handler # Allow ResponseLater to propagate above handler
BaseHandler.old_handle_uncaught_exception = BaseHandler.handle_uncaught_exception BaseHandler.old_handle_uncaught_exception = BaseHandler.handle_uncaught_exception
BaseHandler.handle_uncaught_exception = new_handle_uncaught_exception BaseHandler.handle_uncaught_exception = new_handle_uncaught_exception
def new_handle_uncaught_exception(self, request, resolver, exc_info): def new_handle_uncaught_exception(self, request, resolver, exc_info):
if exc_info[0] is ResponseLater: if exc_info[0] is ResponseLater:
raise raise
return BaseHandler.old_handle_uncaught_exception(self, request, resolver, exc_info) return BaseHandler.old_handle_uncaught_exception(self, request, resolver, exc_info)

120
channels/management/commands/runserver.py
View File

@ -1,60 +1,60 @@
import django import django
import threading import threading
from django.core.management.commands.runserver import Command as RunserverCommand from django.core.management.commands.runserver import Command as RunserverCommand
from django.core.handlers.wsgi import WSGIHandler from django.core.handlers.wsgi import WSGIHandler
from django.http import HttpResponse from django.http import HttpResponse
from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND
from channels.worker import Worker from channels.worker import Worker
from channels.utils import auto_import_consumers from channels.utils import auto_import_consumers
from channels.adapters import UrlConsumer from channels.adapters import UrlConsumer
class Command(RunserverCommand): class Command(RunserverCommand):
def get_handler(self, *args, **options): def get_handler(self, *args, **options):
""" """
Returns the default WSGI handler for the runner. Returns the default WSGI handler for the runner.
""" """
django.setup() django.setup()
return WSGIInterfaceHandler() return WSGIInterfaceHandler()
def run(self, *args, **options): def run(self, *args, **options):
# Force disable reloader for now # Force disable reloader for now
options['use_reloader'] = False options['use_reloader'] = False
# Check a handler is registered for http reqs # Check a handler is registered for http reqs
channel_layer = channel_backends[DEFAULT_CHANNEL_BACKEND] channel_layer = channel_backends[DEFAULT_CHANNEL_BACKEND]
auto_import_consumers() auto_import_consumers()
if not channel_layer.registry.consumer_for_channel("django.wsgi.request"): if not channel_layer.registry.consumer_for_channel("django.wsgi.request"):
# Register the default one # Register the default one
channel_layer.registry.add_consumer(UrlConsumer(), ["django.wsgi.request"]) channel_layer.registry.add_consumer(UrlConsumer(), ["django.wsgi.request"])
# Launch a worker thread # Launch a worker thread
worker = WorkerThread(channel_layer) worker = WorkerThread(channel_layer)
worker.daemon = True worker.daemon = True
worker.start() worker.start()
# Run the rest # Run the rest
return super(Command, self).run(*args, **options) return super(Command, self).run(*args, **options)
class WSGIInterfaceHandler(WSGIHandler): class WSGIInterfaceHandler(WSGIHandler):
""" """
New WSGI handler that pushes requests to channels. New WSGI handler that pushes requests to channels.
""" """
def get_response(self, request): def get_response(self, request):
request.response_channel = Channel.new_name("django.wsgi.response") request.response_channel = Channel.new_name("django.wsgi.response")
Channel("django.wsgi.request").send(**request.channel_encode()) Channel("django.wsgi.request").send(**request.channel_encode())
channel, message = channel_backends[DEFAULT_CHANNEL_BACKEND].receive_many([request.response_channel]) channel, message = channel_backends[DEFAULT_CHANNEL_BACKEND].receive_many([request.response_channel])
return HttpResponse.channel_decode(message) return HttpResponse.channel_decode(message)
class WorkerThread(threading.Thread): class WorkerThread(threading.Thread):
""" """
Class that runs a worker Class that runs a worker
""" """
def __init__(self, channel_layer): def __init__(self, channel_layer):
super(WorkerThread, self).__init__() super(WorkerThread, self).__init__()
self.channel_layer = channel_layer self.channel_layer = channel_layer
def run(self): def run(self):
Worker(channel_layer=self.channel_layer).run() Worker(channel_layer=self.channel_layer).run()

72
channels/request.py
View File

@ -1,36 +1,36 @@
from django.http import HttpRequest from django.http import HttpRequest
from django.utils.datastructures import MultiValueDict from django.utils.datastructures import MultiValueDict
def encode_request(request): def encode_request(request):
""" """
Encodes a request to JSON-compatible datastructures Encodes a request to JSON-compatible datastructures
""" """
# TODO: More stuff # TODO: More stuff
value = { value = {
"GET": request.GET.items(), "GET": request.GET.items(),
"POST": request.POST.items(), "POST": request.POST.items(),
"COOKIES": request.COOKIES, "COOKIES": request.COOKIES,
"META": {k: v for k, v in request.META.items() if not k.startswith("wsgi")}, "META": {k: v for k, v in request.META.items() if not k.startswith("wsgi")},
"path": request.path, "path": request.path,
"path_info": request.path_info, "path_info": request.path_info,
"method": request.method, "method": request.method,
"response_channel": request.response_channel, "response_channel": request.response_channel,
} }
return value return value
def decode_request(value): def decode_request(value):
""" """
Decodes a request JSONish value to a HttpRequest object. Decodes a request JSONish value to a HttpRequest object.
""" """
request = HttpRequest() request = HttpRequest()
request.GET = MultiValueDict(value['GET']) request.GET = MultiValueDict(value['GET'])
request.POST = MultiValueDict(value['POST']) request.POST = MultiValueDict(value['POST'])
request.COOKIES = value['COOKIES'] request.COOKIES = value['COOKIES']
request.META = value['META'] request.META = value['META']
request.path = value['path'] request.path = value['path']
request.method = value['method'] request.method = value['method']
request.path_info = value['path_info'] request.path_info = value['path_info']
request.response_channel = value['response_channel'] request.response_channel = value['response_channel']
return request return request

76
channels/response.py
View File

@ -1,38 +1,38 @@
from django.http import HttpResponse from django.http import HttpResponse
def encode_response(response): def encode_response(response):
""" """
Encodes a response to JSON-compatible datastructures Encodes a response to JSON-compatible datastructures
""" """
# TODO: Entirely useful things like cookies # TODO: Entirely useful things like cookies
value = { value = {
"content_type": getattr(response, "content_type", None), "content_type": getattr(response, "content_type", None),
"content": response.content, "content": response.content,
"status_code": response.status_code, "status_code": response.status_code,
"headers": response._headers.values(), "headers": response._headers.values(),
} }
response.close() response.close()
return value return value
def decode_response(value): def decode_response(value):
""" """
Decodes a response JSONish value to a HttpResponse object. Decodes a response JSONish value to a HttpResponse object.
""" """
response = HttpResponse( response = HttpResponse(
content = value['content'], content = value['content'],
content_type = value['content_type'], content_type = value['content_type'],
status = value['status_code'], status = value['status_code'],
) )
response._headers = {k.lower: (k, v) for k, v in value['headers']} response._headers = {k.lower: (k, v) for k, v in value['headers']}
return response return response
class ResponseLater(Exception): class ResponseLater(Exception):
""" """
Class that represents a response which will be sent doown the response Class that represents a response which will be sent doown the response
channel later. Used to move a django view-based segment onto the next channel later. Used to move a django view-based segment onto the next
task, as otherwise we'd need to write some kind of fake response. task, as otherwise we'd need to write some kind of fake response.
""" """
pass pass

58
channels/utils.py
View File

@ -1,29 +1,29 @@
import types import types
from django.apps import apps from django.apps import apps
def auto_import_consumers(): def auto_import_consumers():
""" """
Auto-import consumers modules in apps Auto-import consumers modules in apps
""" """
for app_config in apps.get_app_configs(): for app_config in apps.get_app_configs():
for submodule in ["consumers", "views"]: for submodule in ["consumers", "views"]:
module_name = "%s.%s" % (app_config.name, submodule) module_name = "%s.%s" % (app_config.name, submodule)
try: try:
__import__(module_name) __import__(module_name)
except ImportError as e: except ImportError as e:
if "no module named %s" % submodule not in str(e).lower(): if "no module named %s" % submodule not in str(e).lower():
raise raise
def name_that_thing(thing): def name_that_thing(thing):
""" """
Returns either the function/class path or just the object's repr Returns either the function/class path or just the object's repr
""" """
if hasattr(thing, "__name__"): if hasattr(thing, "__name__"):
if hasattr(thing, "__class__") and not isinstance(thing, types.FunctionType): if hasattr(thing, "__class__") and not isinstance(thing, types.FunctionType):
if thing.__class__ is not type: if thing.__class__ is not type:
return name_that_thing(thing.__class__) return name_that_thing(thing.__class__)
if hasattr(thing, "__module__"): if hasattr(thing, "__module__"):
return "%s.%s" % (thing.__module__, thing.__name__) return "%s.%s" % (thing.__module__, thing.__name__)
return repr(thing) return repr(thing)

38
channels/worker.py
View File

@ -1,19 +1,19 @@
class Worker(object): class Worker(object):
""" """
A "worker" process that continually looks for available messages to run A "worker" process that continually looks for available messages to run
and runs their consumers. and runs their consumers.
""" """
def __init__(self, channel_layer): def __init__(self, channel_layer):
self.channel_layer = channel_layer self.channel_layer = channel_layer
def run(self): def run(self):
""" """
Tries to continually dispatch messages to consumers. Tries to continually dispatch messages to consumers.
""" """
channels = self.channel_layer.registry.all_channel_names() channels = self.channel_layer.registry.all_channel_names()
while True: while True:
channel, message = self.channel_layer.receive_many(channels) channel, message = self.channel_layer.receive_many(channels)
consumer = self.channel_layer.registry.consumer_for_channel(channel) consumer = self.channel_layer.registry.consumer_for_channel(channel)
consumer(channel=channel, **message) consumer(channel=channel, **message)