mirror
/
amqtt
kopia lustrzana https://github.com/Yakifo/amqtt
1
0
Forkuj 0
Kod Zgłoszenia Packages Projekty Wydania Wiki Aktywność

Merge branch 'rc' into plugin_event_signatures

pull/221/head
Andrew Mirsky 2025-06-16 08:39:14 -04:00
rodzic 3d6e7a8364 eae57bfd8a
commit 3a8c5a64fe
Nie znaleziono w bazie danych klucza dla tego podpisu
ID klucza GPG: A98E67635CDF2C39
1 zmienionych plików z 6 dodań i 1 usunięć
Pokaż statystyki Ściągnij plik aktualizacji Ściągnij plik porównania Expand all files Collapse all files

5
amqtt/plugins/logging_amqtt.py
Wyświetl plik

@ -3,6 +3,7 @@ from functools import partial
import logging import logging
from typing import Any, TypeAlias from typing import Any, TypeAlias
from amqtt.events import BrokerEvents
from amqtt.mqtt import MQTTPacket from amqtt.mqtt import MQTTPacket
from amqtt.mqtt.packet import MQTTFixedHeader, MQTTPayload, MQTTVariableHeader from amqtt.mqtt.packet import MQTTFixedHeader, MQTTPayload, MQTTVariableHeader
from amqtt.plugins.base import BasePlugin from amqtt.plugins.base import BasePlugin
@ -11,13 +12,17 @@ from amqtt.session import Session
PACKET: TypeAlias = MQTTPacket[MQTTVariableHeader, MQTTPayload[MQTTVariableHeader], MQTTFixedHeader] PACKET: TypeAlias = MQTTPacket[MQTTVariableHeader, MQTTPayload[MQTTVariableHeader], MQTTFixedHeader]
class EventLoggerPlugin(BasePlugin[BaseContext]): class EventLoggerPlugin(BasePlugin[BaseContext]):
"""A plugin to log events dynamically based on method names.""" """A plugin to log events dynamically based on method names."""
async def log_event(self, *args: Any, **kwargs: Any) -> None: async def log_event(self, *args: Any, **kwargs: Any) -> None:
"""Log the occurrence of an event.""" """Log the occurrence of an event."""
event_name = kwargs["event_name"].replace("old", "") event_name = kwargs["event_name"].replace("old", "")
if event_name.replace("on_", "") in (BrokerEvents.CLIENT_CONNECTED, BrokerEvents.CLIENT_DISCONNECTED):
self.context.logger.info(f"### '{event_name}' EVENT FIRED ###") self.context.logger.info(f"### '{event_name}' EVENT FIRED ###")
else:
self.context.logger.debug(f"### '{event_name}' EVENT FIRED ###")
def __getattr__(self, name: str) -> Callable[..., Coroutine[Any, Any, None]]: def __getattr__(self, name: str) -> Callable[..., Coroutine[Any, Any, None]]:
"""Dynamically handle calls to methods starting with 'on_'.""" """Dynamically handle calls to methods starting with 'on_'."""