diff --git a/src/netius/base/poll.pyi b/src/netius/base/poll.pyi new file mode 100644 index 00000000..a792709e --- /dev/null +++ b/src/netius/base/poll.pyi @@ -0,0 +1,86 @@ +from typing import Any + +POLL_TIMEOUT: float + +class Poll: + def __init__(self) -> None: ... + @classmethod + def name(cls) -> str: ... + @classmethod + def test(cls) -> bool: ... + def open(self, timeout: float = ...) -> None: ... + def close(self) -> None: ... + def poll(self) -> tuple[list, list, list]: ... + def poll_owner(self) -> dict: ... + def is_open(self) -> bool: ... + def is_edge(self) -> bool: ... + def is_empty(self) -> bool: ... + def sub_all(self, socket, owner: Any | None = ...) -> None: ... + def unsub_all(self, socket) -> None: ... + def is_sub_read(self, socket) -> bool: ... + def is_sub_write(self, socket) -> bool: ... + def is_sub_error(self, socket) -> bool: ... + def sub_read(self, socket, owner: Any | None = ...) -> None: ... + def sub_write(self, socket, owner: Any | None = ...) -> None: ... + def sub_error(self, socket, owner: Any | None = ...) -> None: ... + def unsub_read(self, socket) -> None: ... + def unsub_write(self, socket) -> None: ... + def unsub_error(self, socket) -> None: ... + +class EpollPoll(Poll): + def __init__(self, *args, **kwargs) -> None: ... + @classmethod + def test(cls) -> bool: ... + def open(self, timeout: float = ...) -> None: ... + def close(self) -> None: ... + def poll(self) -> tuple[list, list, list]: ... + def is_edge(self) -> bool: ... + def sub_read(self, socket, owner: Any | None = ...) -> None: ... + def sub_write(self, socket, owner: Any | None = ...) -> None: ... + def sub_error(self, socket, owner: Any | None = ...) -> None: ... + def unsub_read(self, socket) -> None: ... + def unsub_write(self, socket) -> None: ... + def unsub_error(self, socket) -> None: ... + +class KqueuePoll(Poll): + def __init__(self, *args, **kwargs) -> None: ... + @classmethod + def test(cls) -> bool: ... + def open(self, timeout: float = ...) -> None: ... + def close(self) -> None: ... + def poll(self) -> tuple[list, list, list]: ... + def is_edge(self) -> bool: ... + def sub_read(self, socket, owner: Any | None = ...) -> None: ... + def sub_write(self, socket, owner: Any | None = ...) -> None: ... + def sub_error(self, socket, owner: Any | None = ...) -> None: ... + def unsub_read(self, socket) -> None: ... + def unsub_write(self, socket) -> None: ... + def unsub_error(self, socket) -> None: ... + +class PollPoll(Poll): + def __init__(self, *args, **kwargs) -> None: ... + @classmethod + def test(cls) -> bool: ... + def open(self, timeout: float = ...) -> None: ... + def close(self) -> None: ... + def poll(self) -> tuple[list, list, list]: ... + def is_edge(self) -> bool: ... + def sub_read(self, socket, owner: Any | None = ...) -> None: ... + def sub_write(self, socket, owner: Any | None = ...) -> None: ... + def sub_error(self, socket, owner: Any | None = ...) -> None: ... + def unsub_read(self, socket) -> None: ... + def unsub_write(self, socket) -> None: ... + def unsub_error(self, socket) -> None: ... + +class SelectPoll(Poll): + def __init__(self, *args, **kwargs) -> None: ... + def open(self, timeout: float = ...) -> None: ... + def close(self) -> None: ... + def poll(self) -> tuple[list, list, list]: ... + def is_edge(self) -> bool: ... + def sub_read(self, socket, owner: Any | None = ...) -> None: ... + def sub_write(self, socket, owner: Any | None = ...) -> None: ... + def sub_error(self, socket, owner: Any | None = ...) -> None: ... + def unsub_read(self, socket) -> None: ... + def unsub_write(self, socket) -> None: ... + def unsub_error(self, socket) -> None: ...