A consumer is a subclass of either channels.consumer.AsyncConsumer or channels.consumer.SyncConsumer.
Consumers do a couple of things in particular:
- Structures your code as a series of functions to be called whenever an event happens, rather than making you write an event loop.
- Allow you to write synchronous or async code and deals with handoffs and threading for you.
This is another tutorial about Django and channels, you can see the first one.
For testing area you need the postman tool and I install and used with snap tool.
[root@fedora mythcat]# dnf install snapd
Last metadata expiration check: 0:40:03 ago on Fri 14 Jan 2022 03:38:55 PM EET.
...
[root@fedora mythcat]# ln -s /var/lib/snapd/snap /snap
[root@fedora mythcat]# snap install postman
2022-01-14T16:22:15+02:00 INFO Waiting for automatic snapd restart...
postman (v9/stable) 9.8.3 from Postman, Inc. (postman-inc✓) installed
[mythcat@fedora ~]$ snap run postmanLet's go on the project folder:
[mythcat@fedora ~]$ cd djangotest001/
[mythcat@fedora djangotest001]$ cd website001/In this folder I have two folders: appsite001 and website001.
In the appsite001 I add these scripts.
I create a new python script named consumers.py with this source code:
from channels.consumer import SyncConsumer, AsyncConsumer
from channels.exceptions import StopConsumer
class MySyncConsumer(SyncConsumer):
def websocket_connect(self,event):
print('Websocket Connected ...')
self.send({
'type':'websocket.accept',
})
def websocket_receive(self, event):
print('Messaged Received ...')
print(event['text'])
self.send({
'type':'websocket.send',
'text':'Message sent to client'
})
def websocket_diconnect(self, event):
print('Websocket Disconnected ...')
raise StopConsumer
class MyAsyncConsumer(AsyncConsumer):
async def websocket_connect(self,event):
print('Websocket Connected ...')
async def websocket_receive(self, event):
print('Messaged Received ...')
async def websocket_diconnect(self, event):
print('Websocket Disconnected ...')I created routing.py python script with this source code:
from django.urls import path
from . import consumers
websocket_urlpatterns = [
path('ws/sc/',consumers.MySyncConsumer.as_asgi()),
]In the website001 I change this script named asgi.py.
import os
from django.core.asgi import get_asgi_application
from channels.routing import ProtocolTypeRouter, URLRouter
import appsite001.routing
os.environ.setdefault('DJANGO_SETTINGS_MODULE', 'website001.settings')
application = ProtocolTypeRouter({
'http':get_asgi_application(),
'websocket':URLRouter(
appsite001.routing.websocket_urlpatterns
)
})Run the Django project with :
[mythcat@fedora website001]$ python manage.py runserver
Watching for file changes with StatReloader
Performing system checks...
System check identified no issues (0 silenced).
January 14, 2022 - 15:32:29
Django version 4.0.1, using settings 'website001.settings'
Starting ASGI/Channels version 3.0.4 development server at http://127.0.0.1:8000/
Quit the server with CONTROL-C.
WebSocket HANDSHAKING /ws/sc/ [127.0.0.1:33944]
Websocket Connected ...
WebSocket CONNECT /ws/sc/ [127.0.0.1:33944]
Messaged Received ...
This is a message from mythcat
...Use postman tool with websocket to send this message to Django project:
This is a message from mythcatYou can see how this works:
