Method _trigger_event

src/socketio/async_client.py:464–497

Invoke an application event handler.

(self, event, namespace, *args)

            self.connected = False

    async def _trigger_event(self, event, namespace, *args):
        """Invoke an application event handler."""
        # first see if we have an explicit handler for the event
        handler, args = self._get_event_handler(event, namespace, args)
        if handler:
            if inspect.iscoroutinefunction(handler):
                try:
                    try:
                        ret = await handler(*args)
                    except TypeError:
                        # the legacy disconnect event does not take a reason
                        # argument
                        if event == 'disconnect':
                            ret = await handler(*args[:-1])
                        else:  # pragma: no cover
                            raise
                except asyncio.CancelledError:  # pragma: no cover
                    ret = None
            else:
                try:
                    ret = handler(*args)
                except TypeError:
                    # the legacy disconnect event does not take a reason
                    # argument
                    if event == 'disconnect':
                        ret = handler(*args[:-1])
                    else:  # pragma: no cover
                        raise
            return ret

        # or else, forward the event to a namespace handler if one exists
        handler, args = self._get_namespace_handler(namespace, args)
        if handler:
            return await handler.trigger_event(event, *args)

    async def _handle_reconnect(self):
        if self._reconnect_abort is None:  # pragma: no cover