Subscribing to data changes across multiple connections

[zaxarhic2134]

When executing the asyncio.run(main()) command, a for loop occurs in which data is obtained to create a client, and then a subscription to changing variables occurs, but due to the presence of "while True:" the process is blocked and, in fact, only one connection is processed

class SubscriptionHandler:
    def __init__(self, variable):
        self.variable = variable
        print(self.variable)

    async def datachange_notification(self, node: Node, val, data):
        print(node, val)
class OPCUAClient:
    def __init__(self, uri, namespace, connection_id):
        self.client = Client(url=uri)
        self.namespace = namespace
        self.connection_id = connection_id

    async def __aenter__(self):
        await self.client.connect()
        await set_connection_status(self.connection_id, 1)
        return self

    async def __aexit__(self, exc_type, exc_val, exc_tb):
        await self.client.disconnect()
        await set_connection_status(self.connection_id, 2)
async def main():
        for connection in await get_connections():
            try:
                async with OPCUAClient(connection.connection_data['uri'],
                                       connection.connection_data['nsuri'],
                                       connection.id) as OPCClient:
                        client = OPCClient.client
                        nodes = await get_variables(connection.id)
                        for variables in nodes:
                            handler = SubscriptionHandler(variables.id)
                            subscription = await client.create_subscription(100, handler)

                            for nodeId in variables.variables:
                                node = client.get_node(nodeId)
                                handle = await subscription.subscribe_data_change(node)
                        while True:
                            await asyncio.sleep(1)
            except Exception as e:
                await set_connection_status(connection.id, 3)
                print(e)


def run(*args):
    asyncio.run(main())

4
Revised values returned differ from subscription values: CreateSubscriptionResult(SubscriptionId=221, RevisedPublishingInterval=100.0, RevisedLifetimeCount=81000, RevisedMaxKeepAliveCount=27000)
ns=3;i=1006 -2.0
ns=3;i=1004 -0.4
ns=3;i=1004 0.0
ns=3;i=1006 2.0
ns=3;i=1004 0.4

the output is only a subscription for one connection
How can I split the threads so that all connections are processed?

[schroeder-]

Normaly you would run a async task per connection. You can do this with asyncio.create_task and asyncio.gather.

This is how i would do it (code not tested):

class SubscriptionHandler:
     def __init__(self, url, variable):
        self.variable = variable
        self._url = url

     async def datachange_notification(self, node: Node, val, data):
        """
        Callback for asyncua Subscription.
        This method will be called when the Client received a data change message from the Server.
        """
        _logger.info(f'datachange_notification Client: {self._url} : {self._id} {node} {val}')


async def run_loop(connection):
      try:
                async with OPCUAClient(connection.connection_data['uri'],
                                       connection.connection_data['nsuri'],
                                       connection.id) as OPCClient:
                        client = OPCClient.client
                        nodes = await get_variables(connection.id)
                        for variables in nodes:
                            handler = SubscriptionHandler(connection, variables.id)
                            subscription = await client.create_subscription(100, handler)

                            for nodeId in variables.variables:
                                node = client.get_node(nodeId)
                                handle = await subscription.subscribe_data_change(node)
                        while True:
                            await asyncio.sleep(1)
            except Exception as e:
                await set_connection_status(connection.id, 3)
                print(e)

async def main():
        tasks = [asyncio.create_task(run_loop(connection)) for connection in await get_connections()]
        results = await asyncio.gather(*tasks)