Django channels and aiohttp - channels websocket crash & blocking background worker
I am using Django 2.0.5, channels 2.1.1, aiohttp 3.1.3
two issues: (1) everything works for a while and then the channels
websocket dies. Error is 'No handler for message type...'. (2) the
background `await t1` call appears to be blocking and it appears that any
attempt to trigger closeConnection does not work. Would appreciate
help/suggestions with these issues! Code below:
*consumer.py*:
rom channels.generic.websocket import AsyncWebsocketConsumer
import json
class myConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.channel_layer.group_add('myGroup', self.channel_name)
await self.accept()
async def disconnect(self, closeCode):
await self.channel_layer.group_discard('myGroup', self.channel_name)
async def receive(self, text_data):
eventData = json.loads(text_data)
if eventData['monitorToggle'] == 'on':
print('on to background worker.')
await self.channel_layer.send('toggle',{"type":
"openConnection"})
elif eventData['monitorToggle'] == 'off':
print('off to background worker.')
await self.channel_layer.send('toggle',{"type":
"closeConnection"})
async def myConsumerUpdate(self, data):
await self.send(text_data=json.dumps(data['updateTime'))
*backgroundConsumer.py*
from channels.consumer import AsyncConsumer
from channels.layers import get_channel_layer
import asyncio
import aiohttp
import json
from mystuff.models import Info
import datetime
class myConsumer(AsyncConsumer):
turnOnListener = True
channelLayer = get_channel_layer()
async def listen(self):
infoList = Info.objects.filter(active=True)
subInfoList = []
for i in infoList:
subInfoList.append(sometText'+str(i))
async with aiohttp.ClientSession() as session:
async with
session.ws_connect('...streamerURL/socket.io/?EIO=3&transport=websocket',
ssl=False) as ws:
await ws.send_str(str(subInfoList))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
if msg.data.startswith('2'):
await ws.send_str('3')
else:
await self.parseListenData(msg)
async def parseListenData(self, msg):
# some code in here, including call to datetime to parse the msg
updateMsg = {'type': 'myConsumerUpdate',
'updateTime': str(updateTime),
}
print('built update msg and now sending to group')
await self.channelLayer.group_send('marketsGroup', updateMsg)
async def openConnection(self, event):
if(self.turnOnListener):
print('adding group')
await self.channelLayer.group_add(
'myGroup',
self.channel_name,
)
print('turned on listener')
self.turnOnListener = False
loop = asyncio.get_event_loop()
t1 = loop.create_task(self.listen())
print('created t1')
await t1
print('awaiting t1') # Does not print until *consumer.py*
crashes
else:
print('already turned on listener')
async def closeConnection(self, event):
print('in close connection') #Never gets here for some reason due
to `await t1` blocking...
if(not self.turnOnListener):
print('turn it off')
--
You received this message because you are subscribed to the Google Groups
"Django users" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to django-users+unsubscr...@googlegroups.com.
To post to this group, send email to django-users@googlegroups.com.
Visit this group at https://groups.google.com/group/django-users.
To view this discussion on the web visit
https://groups.google.com/d/msgid/django-users/ba76ddb6-eaad-4caf-8547-1473e15a6d9d%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
Django Channels 2.0 websocket crashing & blocking while using aiohttp
I am using Django 2.0.5, Python 3.6.2, Django Channels 2.1.1, aiohttp 3.1.3.
Background:
I am connecting to an external websocket, asynchronously, using aiohttp
within a background channels worker. Once triggered, the socket connection
stays alive, listening/streaming, and sends incoming messages to a consumer
through a group_send. The consumer receives it at the appropriate handler,
however, after some time, it appears that the channels websocket dies and
the handler can no longer be found - I get 'No handler for message
type' even though it has been working successfully until that point.
Second issue is that 'await t1' seems to block in the background. I route
the 'toggle' to backgroundUpdater.py.
*consumers.py*:
from channels.generic.websocket import AsyncWebsocketConsumer
import json
class MyConsumer(AsyncWebsocketConsumer):
async def connect(self):
await self.channel_layer.group_add('myGroup', self.channel_name)
await self.accept()
async def disconnect(self, closeCode):
await self.channel_layer.group_discard('myGroup', self.channel_name)
async def receive(self, text_data):
eventData = json.loads(text_data)
print(eventData)
if eventData['myToggle'] == 'on':
await self.channel_layer.send('toggle',{"type":
"openConnection"})
elif eventData['myToggle'] == 'off':
await self.channel_layer.send('toggle',{"type":
"closeConnection"})
async def backgroundWorkerUpdate(self, data):
await self.send(text_data=json.dumps(data['updateTime']))
*backgroundUpdater.py*:
from channels.consumer import AsyncConsumer
from channels.layers import get_channel_layer
import asyncio
import aiohttp
import json
from dashboard.models import Info
import datetime
class backgroundConsumer(AsyncConsumer):
turnOnListener = True
channelLayer = get_channel_layer()
loadComplete = False
async def listen(self):
infoList = Info.objects.filter(active=True)
subInfoList = []
for i in infoList:
subInfoList.append('info'+str(t))\
self.loadComplete = False
async with aiohttp.ClientSession() as session:
async with
session.ws_connect('...streamingsiteURL.../socket.io/?EIO=3&transport=websocket',
ssl=False) as ws:
await ws.send_str(str(subInfoList))
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
if msg.data.startswith('2'):
await ws.send_str('3')
else:
await self.parseListenData(msg)
async def parseListenData(self, msgContent):
# handles data in here, including a call to datetime
updateMsg = {'type': 'backgroundWorkerUpdate',
'updateTime': str(updateTime),
}
print('built update msg and now sending to group')
await self.channelLayer.group_send('myGroup', updateMsg)
async def openConnection(self, event):
if(self.turnOnListener):
await self.channelLayer.group_add(
'myGroup',
self.channel_name,
)
self.turnOnListener = False
loop = asyncio.get_event_loop()
t1 = loop.create_task(self.listen())
await t1
print('awaiting t1') # THIS NEVER PRINTS UNTIL THE WEBSOCKET IN
consumers.py CRASHES
else:
print('already turned on listener')
async def closeConnection(self, event):
print('in close connection') # CANNOT SEEM TO TRIGGER THIS ONCE
await t1 is called above...
if(not self.turnOnListener):
print('turn it off')
*Questions*:
1. Why is does this work for some time until the websocket in consumers.py
crashes? I am assuming it has crashed because of the missing handler (i.e.
'No handler...' error).
2. Why is the 'await t1' call blocking? When I try to send 'await
self.channel_layer.send('toggleMarket',{"type": "closeConnection"})', it
sends, but nothing seems to be received?
I am have been trying to get this to run in different ways, however, I
cannot seem to resolve these two issues. Any help/suggestions would be
greatly appreciated.
Thanks!
--
You received this message because you are subscribed to the Google Groups
"Django users" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to django-users+unsubscr...@googlegroups.com.
To post to this group, send email to django-users@googlegroups.com.
Visit this group at https://groups.google.com/group/django-users.
To view this discussion on the web visit
https://groups.google.com/d/msgid/django-users/67993058-4e84-4c11-8acd-fd29249a97d6%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
Re: Django Channels 2.0 websocket crashing & blocking while using aiohttp
I'm not quite sure why you seem to be using an AsyncConsumer outside of the
context of the Channels routing framework?
Anyway, it sounds like from what you're describing that your listen()
method is blocking, which makes sense, as it connects to an external
websocket and listens on it for messages forever. Not sure why the crashing
Channels websocket takes down the client one, but maybe they are both made
to crash by some third, external, force?
Andrew
On Sun, Jun 17, 2018 at 11:33 AM Nathan Bluvol
wrote:
> I am using Django 2.0.5, Python 3.6.2, Django Channels 2.1.1, aiohttp
> 3.1.3.
>
> Background:
> I am connecting to an external websocket, asynchronously, using aiohttp
> within a background channels worker. Once triggered, the socket connection
> stays alive, listening/streaming, and sends incoming messages to a consumer
> through a group_send. The consumer receives it at the appropriate handler,
> however, after some time, it appears that the channels websocket dies and
> the handler can no longer be found - I get 'No handler for message
> type' even though it has been working successfully until that point.
> Second issue is that 'await t1' seems to block in the background. I route
> the 'toggle' to backgroundUpdater.py.
>
> *consumers.py*:
>
> from channels.generic.websocket import AsyncWebsocketConsumer
> import json
>
>
> class MyConsumer(AsyncWebsocketConsumer):
>
> async def connect(self):
> await self.channel_layer.group_add('myGroup', self.channel_name)
> await self.accept()
>
> async def disconnect(self, closeCode):
> await self.channel_layer.group_discard('myGroup',
> self.channel_name)
>
> async def receive(self, text_data):
> eventData = json.loads(text_data)
> print(eventData)
> if eventData['myToggle'] == 'on':
> await self.channel_layer.send('toggle',{"type":
> "openConnection"})
> elif eventData['myToggle'] == 'off':
> await self.channel_layer.send('toggle',{"type":
> "closeConnection"})
>
> async def backgroundWorkerUpdate(self, data):
> await self.send(text_data=json.dumps(data['updateTime']))
>
> *backgroundUpdater.py*:
>
> from channels.consumer import AsyncConsumer
> from channels.layers import get_channel_layer
> import asyncio
> import aiohttp
> import json
> from dashboard.models import Info
> import datetime
>
>
> class backgroundConsumer(AsyncConsumer):
> turnOnListener = True
> channelLayer = get_channel_layer()
> loadComplete = False
>
> async def listen(self):
> infoList = Info.objects.filter(active=True)
> subInfoList = []
> for i in infoList:
> subInfoList.append('info'+str(t))\
> self.loadComplete = False
> async with aiohttp.ClientSession() as session:
> async with session.ws_connect('...streamingsiteURL.../
> socket.io/?EIO=3&transport=websocket', ssl=False) as ws:
> await ws.send_str(str(subInfoList))
> async for msg in ws:
> if msg.type == aiohttp.WSMsgType.TEXT:
> if msg.data.startswith('2'):
> await ws.send_str('3')
> else:
> await self.parseListenData(msg)
>
> async def parseListenData(self, msgContent):
> # handles data in here, including a call to datetime
>
> updateMsg = {'type': 'backgroundWorkerUpdate',
> 'updateTime': str(updateTime),
> }
> print('built update msg and now sending to group')
> await self.channelLayer.group_send('myGroup', updateMsg)
>
> async def openConnection(self, event):
> if(self.turnOnListener):
> await self.channelLayer.group_add(
> 'myGroup',
> self.channel_name,
> )
> self.turnOnListener = False
> loop = asyncio.get_event_loop()
> t1 = loop.create_task(self.listen())
> await t1
> print('awaiting t1') # THIS NEVER PRINTS UNTIL THE WEBSOCKET
> IN consumers.py CRASHES
> else:
> print('already turned on listener')
>
> async def closeConnection(self, event):
> print('in close connection') # CANNOT SEEM TO TRIGGER THIS ONCE
> await t1 is called above...
> if(not self.turnOnListener):
> print('turn it off')
>
> *Questions*:
> 1. Why is does this work for some time until the websocket in consumers.py
> crashes? I am assuming it has crashed because of the missing handler (i.e.
> 'No handler...' error).
> 2. Why is the 'await t1' call blocking? When I try to send 'await
> self.channel_layer.send('toggleMarket',{"type": "closeConnection"})', it
> sends, but nothing seems to be received?
>
> I am have been trying to get this to run in different ways, however, I
> cannot seem to resolve these two issues. Any help/suggestions would
Django formset for models without foreign key
I'm trying to use formset to create connections between my Neo4j nodes.
These models have been constructed using the django-neomodel package. They
aren't related using foreign keys, but since this *isn't* an inline formset
that
shouldn't matter, right?
*models.py*
class Person(DjangoNode):
uid = UniqueIdProperty()
name = StringProperty(max_length=50)
created_at = DateTimeProperty(default=datetime.now)
friends = RelationshipTo('Friend', 'FRIENDED', model=FriendRel)
# DjangoNode class must have a Meta class and documentation specifies
'django_node'
class Meta:
app_label = 'django_node'
class FriendRel(DjangoRel):
created_at = DateTimeProperty(default=datetime.now)
class Meta:
app_label = 'django_rel'
*forms.py*
class PersonForm(forms.ModelForm):
class Meta:
model = Person
# don't need fields for automatically assigned keys like `uid` and
`created_at`
fields = ['name']
class FriendRelForm(forms.ModelForm):
class Meta:
model = FriendRel
exclude = () #"creating ModelForm without either 'fields' || 'exclude'
attribute is prohibited"
FriendRelFormSet = formset_factory(FriendRelForm, extra=1)
*form.html*
{% csrf_token %}
{{ form.as_p }}
{{ friends.management_form }}
{% for form in friends.forms %}
{% if forloop.first %}
{% for field in form.visible_fields %}
{{ field.label|capfirst }}
{% endfor %}
{% endif %}
{% for field in form.visible_fields %}
{% if forloop.first %}
{% for hidden in form.hidden_fields %}
{{ hidden }}
{% endfor %}
{% endif %}
{{ field.errors.as_ul }}
{{ field }}
{% endfor %}
{% endfor %}
I'm expecting a "friend" formset to appear in the rendered form, but am not
quite sure how to get it there. If I add the following I get an error:
class FriendRelForm(forms.ModelForm):
class Meta:
model = FriendRel
exclude = ()
fields = ['friends']
ERROR*** django.core.exceptions.FieldError: Unknown field(s) (friends)
specified for FriendRel*
--
You received this message because you are subscribed to the Google Groups
"Django users" group.
To unsubscribe from this group and stop receiving emails from it, send an email
to django-users+unsubscr...@googlegroups.com.
To post to this group, send email to django-users@googlegroups.com.
Visit this group at https://groups.google.com/group/django-users.
To view this discussion on the web visit
https://groups.google.com/d/msgid/django-users/3e387c8a-522b-4278-8784-aff15f0d9602%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.
