Django Channels 2.0 - How to init and persist/communicate with only one instance of background worker for every channel_layer.send?

[itsnate_b]

I am using Django 2.0.5, Python 3.6.2, Django Channels 2.1.1, aiohttp 3.1.3.

I have created a background worker class which is initialized on a route to the appropriate event/function. Problem is that it initializes a *new* instance of the class on every triggered call, meaning that I cannot access contextual variables (they don't persist). Is there a way to trigger the same (i.e. first) and only class repeatedly on each routed call? I can change it all to function calls with global variables stored for the relevant asyncio futures and/or booleans, but would really prefer not to do that....any suggestions?

My (simplified) code:

consumers.py

from channels.generic.websocket import AsyncWebsocketConsumer

from channels.consumer import AsyncConsumer

from channels.layers import get_channel_layer

import json

import asyncio

import aiohttp

import datetime

class mainConsumer(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['toggle'] == 'on':

            await self.channel_layer.send('myToggle',{"type": "openConnection"})

        elif eventData['toggle'] == 'off':

            await self.channel_layer.send('myToggle',{"type": "closeConnection"})

    async def myUpdate(self, data):

        updateMsg = #some data processing here

        await self.send(text_data=json.dumps(updateMsg))

class backgroundConsumer(AsyncConsumer):

    def __init__(self, scope):

        self.scope = scope

        self.turnOnListener = True

        self.channelLayer = get_channel_layer()

        self.t1 = None

        print('inside init') # this will print init on every channel_layer.send to 'myToggle' in mainConsumer

    async def listen(self):

        

        async with aiohttp.ClientSession() as session:

            async with session.ws_connect('...streaming url...', ssl=False) as ws:

                await ws.send_str('...some string...')

                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):
         # process and create updateMsg dict with type: 'myUpdate' (so that it gets to the myUpdate function in main Consumer)

         await self.channelLayer.group_send('myGroup', updateMsg)

    async def openConnection(self, event):

        if(self.turnOnListener): # THIS IS ALWAYS TRUE BECAUSE THE ROUTE INITIALIZES A NEW INSTANCE OF THIS CLASS!!

            await self.channelLayer.group_add(

                'myGroup',

                self.channel_name,

            )

            self.turnOnListener = False # This only changes it for the current instance (obviously)

            loop = asyncio.get_event_loop()

            self.t1 = asyncio.run_coroutine_threadsafe(self.listen(), loop=loop)

        else:

            print('already turned on listener')

    async def closeConnection(self, event):

        loop = asyncio.get_event_loop()

        loop.call_soon_threadsafe(self.t1.cancel) # t1 is always None because the route initialized a new instance of this class!!

routing.py

from channels.auth import AuthMiddlewareStack

from channels.routing import ProtocolTypeRouter, URLRouter, ChannelNameRouter

import dashboard.routing

from dashboard.consumers import backgroundConsumer

application = ProtocolTypeRouter({

    # (http->django views is added by default)

    'websocket': AuthMiddlewareStack(

        URLRouter(

            dashboard.routing.websocket_urlpatterns

        )

    ),

    'channel': ChannelNameRouter({

        "myToggle": backgroundConsumer,

    }),

})

[Andrew Godwin]

The worker server tries to keep one application instance open per incoming channel name, but that's no guarantee. It will always make a new instance for different channel names. What kind of behaviour are you expecting?

Andrew

--
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...@googlegroups.com.
To post to this group, send email to django...@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/431ada54-74cb-4822-9b33-5dcb0326fb9c%40googlegroups.com.
For more options, visit https://groups.google.com/d/optout.

[itsnate_b]

I am only using one channel name and my expectation is that only one instance of the background worker class should be initiated/used. Instead, I am seeing a new class being initiated for each call routed to the background worker class. This is unexpected...

[Andrew Godwin]

I'm not sure why that would happen - the code that chooses what application instance to use is here: https://github.com/django/asgiref/blob/master/asgiref/server.py (under get_or_create_application_instance).

If you're always using a single channel name, which gets passed as scope_id (see here: https://github.com/django/channels/blob/master/channels/worker.py), it really should be returning the same instance. I recommend you poke around in there and do some debugging to figure out what's going on.

Andrew

To view this discussion on the web visit https://groups.google.com/d/msgid/django-users/541378d0-6e7d-43ea-9b08-016aefd38702%40googlegroups.com.