'Event system with event loop.'
from abc import abstractmethod, ABC
+from dataclasses import dataclass
from queue import SimpleQueue, Empty as QueueEmpty
from threading import Thread
from typing import Any, Iterator, Literal, Self
+@dataclass
class Event:
'Communication unit between threads.'
+@dataclass
class AffectiveEvent(Event, ABC):
'For Events that are to affect other objects.'
'To be run by main loop on target.'
+@dataclass
class PayloadMixin:
'Extends with .payload= passed as first argument.'
-
- def __init__(self, payload, **kwargs) -> None:
- super().__init__(**kwargs)
- self.payload = payload
+ payload: Any
+@dataclass
class ExceptionEvent(Event, PayloadMixin):
'To deliver Exception to main loop for handling.'
payload: Exception
+@dataclass
class QuitEvent(Event):
'To break main loop towards.'
+@dataclass
class QueueMixin:
- 'Adds SimpleQueue addressable via ._put(Event).'
-
- def __init__(self, q_out: SimpleQueue, **kwargs) -> None:
- self._q_out = q_out
- super().__init__(**kwargs)
+ 'Adds SimpleQueue .q_out addressable via ._put(Event).'
+ q_out: SimpleQueue
def _put(self, event: Event) -> None:
- self._q_out.put(event)
+ self.q_out.put(event)
class Loop(QueueMixin):
return self._raw
+@dataclass
class ClientIdMixin:
'Collects a Client\'s ID at .client_id.'
-
- def __init__(self, client_id: UUID, **kwargs) -> None:
- super().__init__(**kwargs)
- self.client_id = client_id
+ client_id: UUID
+@dataclass
class NewClientEvent(AffectiveEvent, PayloadMixin):
'Put Client .payload into ClientsDb target.'
payload: 'Client'
target[self.payload.id_] = self.payload
+@dataclass
class ClientEvent(AffectiveEvent, ClientIdMixin):
'To affect Client identified by ClientIdMixin.'
+@dataclass
class _ConnectedEvent(ClientEvent):
def affect(self, target: 'Client') -> None:
target.send(IrcMessage(verb='NICK', params=(target.nickname,)))
+@dataclass
class InitReconnectEvent(ClientEvent):
'To trigger re-opening of connection.'
target.start_connecting()
+@dataclass
class SendEvent(ClientEvent, PayloadMixin):
'To trigger sending of payload to server.'
payload: IrcMessage
target.send(msg=self.payload, chat=self._chat)
+@dataclass
class ClientQueueMixin(QueueMixin):
'To QueueMixin adds _cput to extend ._put with client_id= setting.'
client_id_name = 'id_'
self._socket.settimeout(_TIMEOUT_RECV_LOOP)
self.assumed_open = True
self._recv_loop = Loop(iterator=self._read_lines(),
- q_out=self._q_out)
+ q_out=self.q_out)
self._cput(_ConnectedEvent)
except Exception as e: # pylint: disable=broad-exception-caught
self._put(ExceptionEvent(e))
bytes_total = b''
+@dataclass
class _RecvEvent(ClientEvent, PayloadMixin):
payload: str
from abc import ABC, abstractmethod
from base64 import b64decode
from contextlib import contextmanager
+from dataclasses import dataclass
from inspect import _empty as inspect_empty, signature, stack
from signal import SIGWINCH, signal
from typing import Callable, Generator, Iterator, NamedTuple, Optional
}
+@dataclass
class TuiEvent(AffectiveEvent):
'To affect TUI, and trigger flushed .draw_tainted on it.'
target.term.flush()
+@dataclass
class _SetScreenEvent(TuiEvent):
def affect(self, target: 'Tui') -> None:
widget.draw()
+@dataclass
class _KeyboardEvent(TuiEvent, PayloadMixin):
payload: str
def _new_client_window(self, client_id: UUID, chat: str = ''
) -> '_ClientWindow':
new_idx = len(self.windows)
- win = _ClientWindow(idx=new_idx, term=self.term, q_out=self._q_out,
+ win = _ClientWindow(idx=new_idx, term=self.term, q_out=self.q_out,
client_id=client_id, chat=chat)
self.windows += [win]
self._switch_window(new_idx)
) -> None:
'Create Client and pass it via NewClientEvent.'
self._put(NewClientEvent(
- _ClientKnowingTui(q_out=self._q_out, hostname=hostname,
+ _ClientKnowingTui(q_out=self.q_out, hostname=hostname,
nickname=nickname, realname=realname)))
def cmd__prompt_enter(self) -> None:
with (self._blessed.raw(),
self._blessed.fullscreen(),
self._blessed.hidden_cursor(),
- Loop(iterator=self._get_keypresses(), q_out=self._q_out)):
+ Loop(iterator=self._get_keypresses(), q_out=self.q_out)):
yield self
@property
payload=IrcMessage(verb='PRIVMSG', params=(target, msg)))
+@dataclass
class _ClientWindowEvent(TuiEvent, ClientIdMixin):
-
- def __init__(self, chat: str = '', **kwargs) -> None:
- self.chat = chat
- super().__init__(**kwargs)
+ chat: str = ''
+@dataclass
class _ClientLogEvent(_ClientWindowEvent, PayloadMixin):
payload: str
super().affect(target)
+@dataclass
class _ClientPromptEvent(_ClientWindowEvent, PayloadMixin):
payload: tuple[str, str]