How to handle subscription timeout on unexpected server shutoff.

[Mezilsa]

Hi there,

I need a little help with exception handling.
I just patched from 1.0.2. to 1.0.4 and apperantly the client-watchdog is now working. Yeay! 🥳

Currently I am disconnecting via an unexpected shutdown (aka pausing the vm my OPCUA server is running on to simulate someone pulling the plug) and as expected I get a timeout from the client.

Error in watchdog loop
Traceback (most recent call last):
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\client\client.py", line 520, in _monitor_server_loop
    _ = await self.nodes.server_state.read_value()
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\common\node.py", line 179, in read_value
    result = await self.read_data_value()
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\common\node.py", line 190, in read_data_value
    return await self.read_attribute(ua.AttributeIds.Value, None, raise_on_bad_status)
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\common\node.py", line 304, in read_attribute
    result = await self.session.read(params)
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\client\ua_client.py", line 397, in read
    data = await self.protocol.send_request(request)
  File "D:\Gitlab\FooFolder\venv\lib\site-packages\asyncua\client\ua_client.py", line 160, in send_request
    data = await asyncio.wait_for(self._send_request(request, timeout, message_type), timeout if timeout else None)
  File "C:\Users\Admin\AppData\Local\Programs\Python\Python38-32\lib\asyncio\tasks.py", line 490, in wait_for
    raise exceptions.TimeoutError()
asyncio.exceptions.TimeoutError

And that is all fine and all, but I am a bit...confused where to put the exception handling for this case.
Should I simply create my own client.py and put it there, or is there a different approach for this?
I am sorry if this is a realy basic question, but I'd like to do this properly 'cause watchdog-timeouts are kind of important :D

Thank you for your time and I hope I didn't mess up the formatting, etiquettte or missed the obvious rtfm 🙏

[schroeder-]

See https://github.com/FreeOpcUa/opcua-asyncio/blob/master/examples/client-reconnect.py

[schroeder-]

Because check_connection only checks if _monitor_server_loop throws a exception. If faster response you can change the watchdog_intervall value of the Client. Default is 1.0 which means every second the server status is read to check if the connection exists and to keep the connection alive.

[Mezilsa]

But modifying the 'watchdog_intervall' only influences the time until the error-blob from my initial post occurs.

I feel like I worded my question poorly and will try to do again.

//hic sunt client creation that works
client = Client(url=URL, watchdog_intervall=1.0)
try:
  async with client:
    _logger.warning("Connected")
    while True:
      await asyncio.sleep(1)
      await client.check_connection()  # Throws a exception if connection is lost
except (ConnectionError, ua.UaError):
    _logger.warning("Reconnecting in 2 seconds")
    await asyncio.sleep(2)
except asyncio.exceptions.TimeoutError:
    _logger.warning ("Timeout!")

only handels the timeoutError only before the connection is established and after the connection broke. But it isn't catching the error - blob from my initial post.

I am trying to figure out if it is good practice to write my own asyncua.Client just to squash that timeoutError or if there is a 'cleaner' way of catching it (e.g. ignoring the error blob cause I am handeling it allready because the provided snipped is about the same as my client loop).

Thank you again for your time and patience with me

[Statkern15]

Did you find a solution on how to handle the error blob? I am currently facing the same issue and simply ignoring the message is not something I am fond of.

[Mezilsa]

I...have to admit that I forgor.
I THINK that I simply have stuffed the client in it's own thread and throw every kind of exception handling on it that I could think of.
I'll dump the code bit under here, but I had to prune it quite a bit so it will need some touchups but maybe you can snip a few ideas [and I can get yelled at for doing evil things with code].

No gurantees, takebacks, or complimentary oil change!

Okay, this is too much code

class OPCUA_CLIENT:
    def __init__(self, clientHandler, machine):
        self.machine = machine #data object that does hardware stuff
        self.isBusy = False
        self.isDeleting = False
        self.clientHandler = clientHandler #big and evil class for managing clients
        self.machineName = self.machine.machineName
        self.assignment = None
        self.typeDefinitionsLoaded = False
        self.dataTypeDefinition = {}
        self.eventTypes = {}
        if self.machine is None:
            print("Silly client. Without a machine we can't connect to a machine!")
            return
        self.defaultNodes = {}
        self.robotNodes = {}
        self.client = None
        self.isConnected = False

        # self.userData.notifyListDataChanged.append(self._createTestConnection)
        self.loop = asyncio.new_event_loop()
        asyncio.set_event_loop(self.loop)
        asyncio.run(self.setupClient())

    async def setupClient(self):
        # Set up user authentication
        ipaddress = self.machine.getIpAddress()
        URL = GLOBAL_URL_YOU_DONT_NEED_TO_KNOW
        try:
            self.client = Client(url=URL)
            self.client.session_timeout = 30000
            self.client.application_uri = self.getFilePath("APPLICATION_URI")

            print("Loading User Key Files ", ipaddress)
            await self.client.load_private_key(self.getFilePath("USER_KEY_FILE"))
            await self.client.load_client_certificate(self.getFilePath("USER_CERT_FILE"))
            print("Loading complete ", ipaddress)
            # Set up security profile
            await self.client.set_security(
                policy=SecurityPolicyBasic256Sha256,
                mode=MessageSecurityMode.SignAndEncrypt,
                certificate=self.getFilePath("APP_CERT_FILE"),
                private_key=self.getFilePath("APP_KEY_FILE")
            )

            await asyncio.sleep(1)
            async with self.client:
                await asyncio.sleep(1)
                if not self.typeDefinitionsLoaded:
                    self.dataTypeDefinition.update(await self.client.load_data_type_definitions())
                    self.typeDefinitionsLoaded = True

        except asyncio.exceptions.TimeoutError:
            print("Startup complete, but mashine looks offline.")  # normal error when machine not here Setup is still successful
        except Exception as e:
            print(f"Error connecting to server: {e}")


    def _infinte_subscribtion(self): #called after setup of client is done, handles EVERYTHING
        if self.client is None:
            return False

        loop = asyncio.new_event_loop()
        asyncio.set_event_loop(loop)
        thread = threading.Thread(target=loop.run_until_complete, args=(self.infinte_subscribtion(),))
        thread.start()

    async def infinte_subscribtion(self):
        subscription_handler = SubHandler(self)
        self.client.isBusy = True
        try:
            async with self.client:  # this
                subscription = await self.client.create_subscription(3, subscription_handler)
                namespaceArray = await self.client.get_namespace_array()
                id_core = None
                for namespace in namespaceArray:
                    if namespace.endswith(GLOBAL_URL_YOU_DONT_NEED_TO_KNOW):
                        id_core = await self.client.get_namespace_index(namespace)
                print("Found id_core", id_core)
                print("NC is available, continuing to subscribe")

                self.defaultNodes = await self.findDefaultNodes()
                values_array = list(self.defaultNodes.values())
                values_array.remove(self.defaultNodes["executionStateNode"])
                values_array.remove(self.defaultNodes["channel0ActiveErrorsNode"])

                self.robotNodes = await self.getRobotNodes()
                robotNodes = list(self.robotNodes.values())

                values_array = values_array + robotNodes
                plc_loop_handle = await subscription.subscribe_data_change(values_array)  # each node has a returned handler
                plc_event_handle = await subscription.subscribe_events(machineNode, [
                    "ns=2;i=1004", "ns=2;i=1012"])  # Filter aus den eventTypes holen und nicht mit ID
                try:
                    while True:
                        if self.isDeleting:
                            break

                        if self.getNodesNextCycle:
                            await self.getNotSubscribableNodes() #serverspeficic
                            self.machine.setConnectionStatus(True)
                        else:
                            await asyncio.sleep(1)
                finally:
                    self.machine.setConnectionStatus(False)
                    for handle in plc_loop_handle:
                        if isinstance(handle, int):
                            await subscription.unsubscribe(handle)
                    await subscription.unsubscribe(plc_event_handle)

        except asyncio.TimeoutError as e:
            self.machine.setConnectionStatus(False)
            pass
        except ua.UaStatusCodeError as e:
            if e.code == ua.StatusCodes.BadLicenseNotAvailable:
                print("Licence is not available.;")
            else:
                print("Unhandled UA Error: ", e.code)
                self.machine.setConnectionStatus(False)
        except OSError as e:
            print("OS Fehler, outer Try", e)
            self.connectionErrorCount = self.connectionErrorCount + 1
            if self.connectionErrorCount > 3:
                self.clientHandler.restartClient(self)
                print("REMOVING Client from machine", self.machineName)
        except ConnectionRefusedError:
            print("Windows ist silly and I don't know why.")
            self.machine.setConnectionStatus(False)
        except ConnectionError:
            print("Connection is closed, Errorcount", self.connectionErrorCount)
            self.connectionErrorCount = self.connectionErrorCount + 1
            if self.connectionErrorCount > 3:
                self.clientHandler.restartClient(self)
                print("REMOVING Client from machine", self.machineName)
            # error on VM restart. We need a new Client because Reasons.

class SubHandler:
    def __init__(self, client):
        self.client = client
        self.machine = client.machine

    """
    Subscription Handler. To receive events from server for a subscription
    data_change and event methods are called directly from receiving thread.
    Do not do expensive, slow or network base operations there. Create another
    thread if you need to do such a thing
    """

    def datachange_notification(self, node, value, data):
        """
        Callback method for data subscription event.
        """
        self.client.isConnected = True
        self.opmodeDefinition = None
        print ("here was a lot of code and now it isn'T")

    def status_change_notification(self, status):
        print("Well that is a status change", status)
        self.client.isConnected = False
        self.client.isBusy = False

    def event_notification(self, event):
         print ("todo", event)
         pass

[Statkern15]

Thanks for the response and the code snipped. I will take that as a startpoint to figure out a solution for my problem.