diff options
Diffstat (limited to 'ouroboros/event.py')
| -rw-r--r-- | ouroboros/event.py | 114 |
1 files changed, 78 insertions, 36 deletions
diff --git a/ouroboros/event.py b/ouroboros/event.py index aa03ad4..24b58b8 100644 --- a/ouroboros/event.py +++ b/ouroboros/event.py @@ -1,30 +1,16 @@ # -# Ouroboros - Copyright (C) 2016 - 2026 -# -# Python API for Ouroboros -# -# Dimitri Staessens <dimitri@ouroboros.rocks> -# -# This library is free software; you can redistribute it and/or -# modify it under the terms of the GNU Lesser General Public License -# version 2.1 as published by the Free Software Foundation. -# -# This library is distributed in the hope that it will be useful, -# but WITHOUT ANY WARRANTY; without even the implied warranty of -# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the GNU -# Lesser General Public License for more details. -# -# You should have received a copy of the GNU Lesser General Public -# License along with this library; if not, write to the Free Software -# Foundation, Inc., http://www.fsf.org/about/contact/. +# SPDX-FileCopyrightText: 2016 - 2026 Dimitri Staessens +# SPDX-License-Identifier: LGPL-2.1-only # -"""Asynchronous flow event monitoring for Ouroboros.""" +""" +Asynchronous flow event monitoring for Ouroboros. +""" from __future__ import annotations from enum import IntFlag -from typing import List, Optional, Tuple +from typing import TYPE_CHECKING, Any, List, Optional, Tuple, Type from _ouroboros_dev_cffi import ffi, lib @@ -32,9 +18,15 @@ from ouroboros._timespec import fl_to_timespec from ouroboros.dev import Flow from ouroboros.errors import FlowEventError, raise_errno +if TYPE_CHECKING: + from types import TracebackType + class FEventType(IntFlag): - """Types of flow events.""" + """ + Types of flow events. + """ + FLOW_PKT = lib.FLOW_PKT FLOW_DOWN = lib.FLOW_DOWN FLOW_UP = lib.FLOW_UP @@ -44,7 +36,9 @@ class FEventType(IntFlag): class FEventQueue: - """A queue of flow events waiting to be drained.""" + """ + A queue of flow events waiting to be drained. + """ def __init__(self) -> None: self._fq = lib.fqueue_create() @@ -54,40 +48,58 @@ class FEventQueue: def __enter__(self) -> FEventQueue: return self - def __exit__(self, exc_type, exc_value, tb) -> None: + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: self.destroy() def __del__(self) -> None: self.destroy() def destroy(self) -> None: - """Destroy the underlying queue. Idempotent.""" + """ + Destroy the underlying queue. Idempotent. + """ + if self._fq != ffi.NULL: lib.fqueue_destroy(self._fq) + self._fq = ffi.NULL def next(self) -> Tuple[Flow, FEventType]: - """Return the next ``(flow, event_type)`` pair from the queue.""" + """ + Return the next ``(flow, event_type)`` pair from the queue. + + :return: The :class:`~ouroboros.dev.Flow` the event occurred + on, and its :class:`FEventType`. + :raises FlowEventError: On a negative ouroboros return code. + """ + fd = lib.fqueue_next(self._fq) + raise_errno(fd, default=FlowEventError) _type = lib.fqueue_type(self._fq) + raise_errno(_type, default=FlowEventError) return Flow(fd), FEventType(_type) @property - def handle(self): - """Return the underlying ``fqueue_t *``. + def handle(self) -> Any: + """ + Return the underlying ``fqueue_t *``. - Intended for use by :meth:`FlowSet.wait` within the same - package; not part of the user-facing API. + :return: The underlying ``fqueue_t *`` CFFI pointer. """ + return self._fq class FlowSet: - """A set of flows that can be monitored for events.""" + """ + A set of flows that can be monitored for events. + """ def __init__(self, flows: Optional[List[Flow]] = None) -> None: self._set = lib.fset_create() @@ -98,26 +110,40 @@ class FlowSet: for flow in flows: if lib.fset_add(self._set, flow.fileno()) != 0: lib.fset_destroy(self._set) + self._set = ffi.NULL raise MemoryError(f"Failed to add flow {flow.fileno()}.") def __enter__(self) -> FlowSet: return self - def __exit__(self, exc_type, exc_value, tb) -> None: + def __exit__(self, exc_type: Optional[Type[BaseException]], + exc_value: Optional[BaseException], + tb: Optional[TracebackType]) -> None: self.destroy() def __del__(self) -> None: self.destroy() def destroy(self) -> None: - """Destroy the underlying set. Idempotent.""" + """ + Destroy the underlying set. Idempotent. + """ + if self._set != ffi.NULL: lib.fset_destroy(self._set) + self._set = ffi.NULL def add(self, flow: Flow) -> None: - """Add *flow* to this set.""" + """ + Add *flow* to this set. + + :param flow: The flow to add to this set. + :raises ValueError: If the FlowSet has been destroyed. + :raises MemoryError: If the flow could not be added. + """ + if self._set == ffi.NULL: raise ValueError("FlowSet has been destroyed") @@ -125,14 +151,25 @@ class FlowSet: raise MemoryError(f"Failed to add flow {flow.fileno()}") def zero(self) -> None: - """Remove all flows from this set.""" + """ + Remove all flows from this set. + + :raises ValueError: If the FlowSet has been destroyed. + """ + if self._set == ffi.NULL: raise ValueError("FlowSet has been destroyed") lib.fset_zero(self._set) def remove(self, flow: Flow) -> None: - """Remove *flow* from this set.""" + """ + Remove *flow* from this set. + + :param flow: The flow to remove from this set. + :raises ValueError: If the FlowSet has been destroyed. + """ + if self._set == ffi.NULL: raise ValueError("FlowSet has been destroyed") @@ -141,15 +178,20 @@ class FlowSet: def wait(self, fq: FEventQueue, timeo: Optional[float] = None) -> None: - """Wait for at least one event on one of the monitored flows. + """ + Wait for at least one event on one of the monitored flows. :param fq: The :class:`FEventQueue` to drain events into. :param timeo: Wait timeout in seconds (None blocks forever). + :raises ValueError: If the FlowSet has been destroyed. + :raises FlowEventError: On a negative ouroboros return code. """ + if self._set == ffi.NULL: raise ValueError("FlowSet has been destroyed") _timeo = fl_to_timespec(ffi, timeo) rc = lib.fevent(self._set, fq.handle, _timeo) + raise_errno(rc, default=FlowEventError) |