Merge branch 'master' of github.com:andrewgodwin/channels into doc-update

This commit is contained in:
Faris Chebib 2015-09-10 15:35:48 -06:00
commit f4cb5864e1
10 changed files with 163 additions and 214 deletions

View File

@ -1,4 +1,4 @@
__version__ = "0.1.1" __version__ = "0.8"
# Load backends, using settings if available (else falling back to a default) # Load backends, using settings if available (else falling back to a default)
DEFAULT_CHANNEL_BACKEND = "default" DEFAULT_CHANNEL_BACKEND = "default"

View File

@ -1,97 +1,9 @@
import asyncio import asyncio
import django
import time import time
from autobahn.asyncio.websocket import WebSocketServerProtocol, WebSocketServerFactory from autobahn.asyncio.websocket import WebSocketServerProtocol, WebSocketServerFactory
from collections import deque
from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND from .websocket_autobahn import get_protocol, get_factory
class InterfaceProtocol(WebSocketServerProtocol):
"""
Protocol which supports WebSockets and forwards incoming messages to
the websocket channels.
"""
def onConnect(self, request):
self.channel_backend = channel_backends[DEFAULT_CHANNEL_BACKEND]
self.request_info = {
"path": request.path,
"get": request.params,
}
def onOpen(self):
# Make sending channel
self.reply_channel = Channel.new_name("!websocket.send")
self.request_info["reply_channel"] = self.reply_channel
self.last_keepalive = time.time()
self.factory.protocols[self.reply_channel] = self
# Send news that this channel is open
Channel("websocket.connect").send(self.request_info)
def onMessage(self, payload, isBinary):
if isBinary:
Channel("websocket.receive").send(dict(
self.request_info,
content = payload,
binary = True,
))
else:
Channel("websocket.receive").send(dict(
self.request_info,
content = payload.decode("utf8"),
binary = False,
))
def serverSend(self, content, binary=False, **kwargs):
"""
Server-side channel message to send a message.
"""
if binary:
self.sendMessage(content, binary)
else:
self.sendMessage(content.encode("utf8"), binary)
def serverClose(self):
"""
Server-side channel message to close the socket
"""
self.sendClose()
def onClose(self, wasClean, code, reason):
if hasattr(self, "reply_channel"):
del self.factory.protocols[self.reply_channel]
Channel("websocket.disconnect").send(self.request_info)
def sendKeepalive(self):
"""
Sends a keepalive packet on the keepalive channel.
"""
Channel("websocket.keepalive").send(self.request_info)
self.last_keepalive = time.time()
class InterfaceFactory(WebSocketServerFactory):
"""
Factory which keeps track of its open protocols' receive channels
and can dispatch to them.
"""
# TODO: Clean up dead protocols if needed?
def __init__(self, *args, **kwargs):
super(InterfaceFactory, self).__init__(*args, **kwargs)
self.protocols = {}
def reply_channels(self):
return self.protocols.keys()
def dispatch_send(self, channel, message):
if message.get("close", False):
self.protocols[channel].serverClose()
else:
self.protocols[channel].serverSend(**message)
class WebsocketAsyncioInterface(object): class WebsocketAsyncioInterface(object):
@ -106,8 +18,8 @@ class WebsocketAsyncioInterface(object):
self.port = port self.port = port
def run(self): def run(self):
self.factory = InterfaceFactory("ws://0.0.0.0:%i" % self.port, debug=False) self.factory = get_factory(WebSocketServerFactory)("ws://0.0.0.0:%i" % self.port, debug=False)
self.factory.protocol = InterfaceProtocol self.factory.protocol = get_protocol(WebSocketServerProtocol)
self.loop = asyncio.get_event_loop() self.loop = asyncio.get_event_loop()
coro = self.loop.create_server(self.factory, '0.0.0.0', 9000) coro = self.loop.create_server(self.factory, '0.0.0.0', 9000)
server = self.loop.run_until_complete(coro) server = self.loop.run_until_complete(coro)
@ -125,6 +37,8 @@ class WebsocketAsyncioInterface(object):
""" """
Run in a separate thread; reads messages from the backend. Run in a separate thread; reads messages from the backend.
""" """
# Wait for main loop to start
time.sleep(0.5)
while True: while True:
channels = self.factory.reply_channels() channels = self.factory.reply_channels()
# Quit if reactor is stopping # Quit if reactor is stopping

View File

@ -0,0 +1,101 @@
import time
from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND
def get_protocol(base):
class InterfaceProtocol(base):
"""
Protocol which supports WebSockets and forwards incoming messages to
the websocket channels.
"""
def onConnect(self, request):
self.channel_backend = channel_backends[DEFAULT_CHANNEL_BACKEND]
self.request_info = {
"path": request.path,
"get": request.params,
}
def onOpen(self):
# Make sending channel
self.reply_channel = Channel.new_name("!websocket.send")
self.request_info["reply_channel"] = self.reply_channel
self.last_keepalive = time.time()
self.factory.protocols[self.reply_channel] = self
# Send news that this channel is open
Channel("websocket.connect").send(self.request_info)
def onMessage(self, payload, isBinary):
if isBinary:
Channel("websocket.receive").send({
"reply_channel": self.reply_channel,
"content": payload,
"binary": True,
})
else:
Channel("websocket.receive").send({
"reply_channel": self.reply_channel,
"content": payload.decode("utf8"),
"binary": False,
})
def serverSend(self, content, binary=False, **kwargs):
"""
Server-side channel message to send a message.
"""
if binary:
self.sendMessage(content, binary)
else:
self.sendMessage(content.encode("utf8"), binary)
def serverClose(self):
"""
Server-side channel message to close the socket
"""
self.sendClose()
def onClose(self, wasClean, code, reason):
if hasattr(self, "reply_channel"):
del self.factory.protocols[self.reply_channel]
Channel("websocket.disconnect").send({
"reply_channel": self.reply_channel,
})
def sendKeepalive(self):
"""
Sends a keepalive packet on the keepalive channel.
"""
Channel("websocket.keepalive").send({
"reply_channel": self.reply_channel,
})
self.last_keepalive = time.time()
return InterfaceProtocol
def get_factory(base):
class InterfaceFactory(base):
"""
Factory which keeps track of its open protocols' receive channels
and can dispatch to them.
"""
# TODO: Clean up dead protocols if needed?
def __init__(self, *args, **kwargs):
super(InterfaceFactory, self).__init__(*args, **kwargs)
self.protocols = {}
def reply_channels(self):
return self.protocols.keys()
def dispatch_send(self, channel, message):
if message.get("close", False):
self.protocols[channel].serverClose()
else:
self.protocols[channel].serverSend(**message)
return InterfaceFactory

View File

@ -1,97 +1,9 @@
import django
import time import time
from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory from autobahn.twisted.websocket import WebSocketServerProtocol, WebSocketServerFactory
from collections import deque
from twisted.internet import reactor from twisted.internet import reactor
from channels import Channel, channel_backends, DEFAULT_CHANNEL_BACKEND from .websocket_autobahn import get_protocol, get_factory
class InterfaceProtocol(WebSocketServerProtocol):
"""
Protocol which supports WebSockets and forwards incoming messages to
the websocket channels.
"""
def onConnect(self, request):
self.channel_backend = channel_backends[DEFAULT_CHANNEL_BACKEND]
self.request_info = {
"path": request.path,
"get": request.params,
}
def onOpen(self):
# Make sending channel
self.reply_channel = Channel.new_name("!websocket.send")
self.request_info["reply_channel"] = self.reply_channel
self.last_keepalive = time.time()
self.factory.protocols[self.reply_channel] = self
# Send news that this channel is open
Channel("websocket.connect").send(self.request_info)
def onMessage(self, payload, isBinary):
if isBinary:
Channel("websocket.receive").send(dict(
self.request_info,
content = payload,
binary = True,
))
else:
Channel("websocket.receive").send(dict(
self.request_info,
content = payload.decode("utf8"),
binary = False,
))
def serverSend(self, content, binary=False, **kwargs):
"""
Server-side channel message to send a message.
"""
if binary:
self.sendMessage(content, binary)
else:
self.sendMessage(content.encode("utf8"), binary)
def serverClose(self):
"""
Server-side channel message to close the socket
"""
self.sendClose()
def onClose(self, wasClean, code, reason):
if hasattr(self, "reply_channel"):
del self.factory.protocols[self.reply_channel]
Channel("websocket.disconnect").send(self.request_info)
def sendKeepalive(self):
"""
Sends a keepalive packet on the keepalive channel.
"""
Channel("websocket.keepalive").send(self.request_info)
self.last_keepalive = time.time()
class InterfaceFactory(WebSocketServerFactory):
"""
Factory which keeps track of its open protocols' receive channels
and can dispatch to them.
"""
# TODO: Clean up dead protocols if needed?
def __init__(self, *args, **kwargs):
super(InterfaceFactory, self).__init__(*args, **kwargs)
self.protocols = {}
def reply_channels(self):
return self.protocols.keys()
def dispatch_send(self, channel, message):
if message.get("close", False):
self.protocols[channel].serverClose()
else:
self.protocols[channel].serverSend(**message)
class WebsocketTwistedInterface(object): class WebsocketTwistedInterface(object):
@ -106,8 +18,8 @@ class WebsocketTwistedInterface(object):
self.port = port self.port = port
def run(self): def run(self):
self.factory = InterfaceFactory("ws://0.0.0.0:%i" % self.port, debug=False) self.factory = get_factory(WebSocketServerFactory)("ws://0.0.0.0:%i" % self.port, debug=False)
self.factory.protocol = InterfaceProtocol self.factory.protocol = get_protocol(WebSocketServerProtocol)
reactor.listenTCP(self.port, self.factory) reactor.listenTCP(self.port, self.factory)
reactor.callInThread(self.backend_reader) reactor.callInThread(self.backend_reader)
reactor.callLater(1, self.keepalive_sender) reactor.callLater(1, self.keepalive_sender)

View File

@ -65,9 +65,9 @@ you can write a function to consume a channel, like so::
def my_consumer(message): def my_consumer(message):
pass pass
And then assign a channel to it like this in the channel backend settings:: And then assign a channel to it like this in the channel routing::
"ROUTING": { channel_routing = {
"some-channel": "myapp.consumers.my_consumer", "some-channel": "myapp.consumers.my_consumer",
} }

View File

@ -44,21 +44,25 @@ For now, we want to override the *channel routing* so that, rather than going
to the URL resolver and our normal view stack, all HTTP requests go to our to the URL resolver and our normal view stack, all HTTP requests go to our
custom consumer we wrote above. Here's what that looks like:: custom consumer we wrote above. Here's what that looks like::
# In settings.py
CHANNEL_BACKENDS = { CHANNEL_BACKENDS = {
"default": { "default": {
"BACKEND": "channels.backends.database.DatabaseChannelBackend", "BACKEND": "channels.backends.database.DatabaseChannelBackend",
"ROUTING": { "ROUTING": "myproject.routing.channel_routing",
"http.request": "myproject.myapp.consumers.http_consumer",
},
}, },
} }
# In routing.py
channel_routing = {
"http.request": "myproject.myapp.consumers.http_consumer",
}
As you can see, this is a little like Django's ``DATABASES`` setting; there are As you can see, this is a little like Django's ``DATABASES`` setting; there are
named channel backends, with a default one called ``default``. Each backend named channel backends, with a default one called ``default``. Each backend
needs a class specified which powers it - we'll come to the options there later - needs a class specified which powers it - we'll come to the options there later -
and a routing scheme, which can either be defined directly as a dict or as and a routing scheme, which points to a dict containing the routing settings.
a string pointing to a dict in another file (if you'd rather keep it outside It's recommended you call this ``routing.py`` and put it alongside ``urls.py``
settings). in your project.
If you start up ``python manage.py runserver`` and go to If you start up ``python manage.py runserver`` and go to
``http://localhost:8000``, you'll see that, rather than a default Django page, ``http://localhost:8000``, you'll see that, rather than a default Django page,
@ -78,13 +82,8 @@ serve HTTP requests from now on - and make this WebSocket consumer instead::
Hook it up to the ``websocket.connect`` channel like this:: Hook it up to the ``websocket.connect`` channel like this::
CHANNEL_BACKENDS = { channel_routing = {
"default": { "websocket.connect": "myproject.myapp.consumers.ws_add",
"BACKEND": "channels.backends.database.DatabaseChannelBackend",
"ROUTING": {
"websocket.connect": "myproject.myapp.consumers.ws_add",
},
},
} }
Now, let's look at what this is doing. It's tied to the Now, let's look at what this is doing. It's tied to the
@ -116,12 +115,10 @@ group it's not in)::
Of course, this is exactly the same code as the ``connect`` handler, so let's Of course, this is exactly the same code as the ``connect`` handler, so let's
just route both channels to the same consumer:: just route both channels to the same consumer::
... channel_routing = {
"ROUTING": {
"websocket.connect": "myproject.myapp.consumers.ws_add", "websocket.connect": "myproject.myapp.consumers.ws_add",
"websocket.keepalive": "myproject.myapp.consumers.ws_add", "websocket.keepalive": "myproject.myapp.consumers.ws_add",
}, }
...
And, even though channels will expire out, let's add an explicit ``disconnect`` And, even though channels will expire out, let's add an explicit ``disconnect``
handler to clean up as people disconnect (most channels will cleanly disconnect handler to clean up as people disconnect (most channels will cleanly disconnect
@ -152,18 +149,13 @@ any message sent in to all connected clients. Here's all the code::
def ws_disconnect(message): def ws_disconnect(message):
Group("chat").discard(message.reply_channel) Group("chat").discard(message.reply_channel)
And what our routing should look like in ``settings.py``:: And what our routing should look like in ``routing.py``::
CHANNEL_BACKENDS = { channel_routing = {
"default": { "websocket.connect": "myproject.myapp.consumers.ws_add",
"BACKEND": "channels.backends.database.DatabaseChannelBackend", "websocket.keepalive": "myproject.myapp.consumers.ws_add",
"ROUTING": { "websocket.receive": "myproject.myapp.consumers.ws_message",
"websocket.connect": "myproject.myapp.consumers.ws_add", "websocket.disconnect": "myproject.myapp.consumers.ws_disconnect",
"websocket.keepalive": "myproject.myapp.consumers.ws_add",
"websocket.receive": "myproject.myapp.consumers.ws_message",
"websocket.disconnect": "myproject.myapp.consumers.ws_disconnect",
},
},
} }
With all that code in your ``consumers.py`` file, you now have a working With all that code in your ``consumers.py`` file, you now have a working

View File

@ -31,3 +31,4 @@ Contents:
scaling scaling
backends backends
faqs faqs
releases/index

22
docs/releases/0.8.rst Normal file
View File

@ -0,0 +1,22 @@
0.8 (2015-09-10)
----------------
This release reworks a few of the core concepts to make the channel layer
more efficient and user friendly:
* Channel names now do not start with ``django``, and are instead just ``http.request``, etc.
* HTTP headers/GET/etc are only sent with ``websocket.connect`` rather than all websocket requests,
to save a lot of bandwidth in the channel layer.
* The session/user decorators were renamed, and a ``@channel_session_user`` and ``transfer_user`` set of functions
added to allow moving the user details from the HTTP session to the channel session in the ``connect`` consumer.
* A ``@linearize`` decorator was added to help ensure a ``connect``/``receive`` pair are not executed
simultanously on two different workers.
* Channel backends gained locking mechanisms to support the ``linearize`` feature.
* ``runwsserver`` will use asyncio rather than Twisted if it's available.
* Message formats have been made a bit more consistent.

7
docs/releases/index.rst Normal file
View File

@ -0,0 +1,7 @@
Release Notes
-------------
.. toctree::
:maxdepth: 1
0.8

View File

@ -2,7 +2,7 @@ from setuptools import find_packages, setup
setup( setup(
name='channels', name='channels',
version="0.7", version="0.8",
url='http://github.com/andrewgodwin/django-channels', url='http://github.com/andrewgodwin/django-channels',
author='Andrew Godwin', author='Andrew Godwin',
author_email='andrew@aeracode.org', author_email='andrew@aeracode.org',