-
-
Save jaishirb/ea7e1660146c5ac2a15f5e0cd7aa9631 to your computer and use it in GitHub Desktop.
channels 2 example
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
import json | |
from asgiref.sync import async_to_sync | |
from channels.generic.websocket import WebsocketConsumer | |
from apps.utils.print_colors import _green | |
class UserRequestConsumer(WebsocketConsumer): | |
def connect(self): | |
self.user_id = self.scope['url_route']['kwargs']['user_id'] | |
self.channel_identifier = 'requests_{}'.format(self.user_id) | |
async_to_sync(self.channel_layer.group_add)( | |
self.channel_identifier, | |
self.channel_name | |
) | |
self.accept() | |
def disconnect(self, close_code): | |
async_to_sync(self.channel_layer.group_discard)( | |
self.channel_identifier, | |
self.channel_name | |
) | |
def receive(self, text): | |
text_data_json = json.loads(text) | |
message = text_data_json['message'] | |
async_to_sync(self.channel_layer.group_send)( | |
self.channel_identifier, | |
{ | |
'type': 'request_message', | |
'message': message | |
} | |
) | |
def request_message(self, event): | |
message = event['message'] | |
self.send(text_data=json.dumps({ | |
'message': message | |
})) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from asgiref.sync import async_to_sync | |
from channels.layers import get_channel_layer | |
class UserRequest(Osw4lModel): | |
status = models.CharField( | |
max_length=20, | |
choices=REQUEST_STATES, | |
default=FIRST | |
) | |
new = models.BooleanField(default=True) | |
class Meta: | |
verbose_name = 'User Request' | |
verbose_name_plural = 'Users Requests' | |
def get_notification_data(self, new): | |
return { | |
'id': self.id, | |
'status': self.status, | |
'new': new | |
} | |
def send_notification(self, new): | |
notification = self.get_notification_data(new=new) | |
layer = get_channel_layer() | |
async_to_sync(layer.group_send)('requests_5', { | |
'type': 'request_message', | |
'message': json.dumps(notification) | |
}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from channels.auth import AuthMiddlewareStack | |
from channels.routing import ProtocolTypeRouter, URLRouter | |
import apps.solicitudes.routing | |
application = ProtocolTypeRouter({ | |
"websocket": AuthMiddlewareStack( | |
URLRouter( | |
apps.solicitudes.routing.websocket_urlpatterns | |
) | |
), | |
}) |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
from django.urls import re_path | |
from . import consumers | |
websocket_urlpatterns = [ | |
re_path(r'ws/requests/(?P<user_id>\w+)/$', consumers.UserRequestConsumer), | |
] |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment