from threading import Thread
from typing import Callable, Iterator, NamedTuple, Optional, Self
# ourselves
-from ircplom.events import (BroadcastMixin, Event, EventQueue, Loop,
- PayloadMixin)
+from ircplom.events import BroadcastMixin, Event, Loop, PayloadMixin
TIMEOUT_LOOP = 0.1
return self._raw
-class _ConnIdxMixin:
- 'Collects a Connection ID at .conn_idx.'
-
- def __init__(self, conn_idx: int, **kwargs) -> None:
- super().__init__(**kwargs)
- self.conn_idx = conn_idx
-
-
@dataclass
class LoginNames:
'Collects the names needed on server connect for USER, NICK commands.'
nick_confirmed: bool = False
+class ConnMixin:
+ 'Collects an IrcConnection at .conn.'
+
+ def __init__(self, conn: 'IrcConnection', **kwargs) -> None:
+ super().__init__(**kwargs)
+ self.conn = conn
+
+
class InitConnectEvent(Event, PayloadMixin):
'Event to trigger connection, with payload (host, LoginNames).'
payload: tuple[str, LoginNames]
-class ConnEvent(Event, _ConnIdxMixin):
- 'Event with .conn_idx.'
-
-
-class InitConnWindowEvent(ConnEvent, PayloadMixin):
- 'Event to trigger TUI making ConnectionWindow.'
- payload: LoginNames
+class ConnEvent(Event, ConnMixin):
+ 'Event with .conn.'
class _ConnectedEvent(ConnEvent):
class _SendEvent(ConnEvent, PayloadMixin):
'Event to trigger sending of payload to server.'
- payload: '_IrcMessage'
-
-
-class BroadcastConnMixin(BroadcastMixin, _ConnIdxMixin):
- 'Provides .broadcast_conn on classes that have .conn_idx defined.'
+ payload: _IrcMessage
- def broadcast_conn[E: ConnEvent](self,
- event_class: type[E],
- *args, **kwargs
- ) -> None:
- 'Broadcast event subclassing ConnEvent, with connection ID.'
- self.broadcast(event_class, conn_idx=self.conn_idx, *args, **kwargs)
- def send(self, verb: str, parameters: tuple[str, ...]) -> None:
- 'Broadcast _SendEvent for _IrcMessage(verb, parameters).'
- self.broadcast_conn(_SendEvent, _IrcMessage(verb, parameters))
-
-
-class IrcConnection(BroadcastConnMixin):
+class IrcConnection(BroadcastMixin):
'Abstracts socket connection, loop over it, and handling messages from it.'
def __init__(self,
- q_to_main: EventQueue,
- idx: int,
hostname: str,
login: LoginNames,
+ **kwargs
) -> None:
- super().__init__(conn_idx=idx, q_to_main=q_to_main)
+ super().__init__(**kwargs)
self._hostname = hostname
- self._login = login
+ self.login = login
self._socket: Optional[socket] = None
self._assumed_open = False
self._recv_loop: Optional[_RecvLoop] = None
- self.broadcast_conn(InitConnWindowEvent, self._login)
self._start_connecting()
def _start_connecting(self) -> None:
def connect(self) -> None:
self._socket = socket()
- self.broadcast_conn(LogConnEvent,
- f'Connecting to {self._hostname} …')
+ self.broadcast(LogConnEvent,
+ f'Connecting to {self._hostname} …')
self._socket.settimeout(_TIMEOUT_CONNECT)
try:
self._socket.connect((self._hostname, _PORT))
except (TimeoutError, socket_gaierror) as e:
- self.broadcast_conn(LogConnEvent, f'ALERT: {e}')
+ self.broadcast(LogConnEvent, f'ALERT: {e}')
return
self._socket.settimeout(TIMEOUT_LOOP)
self._assumed_open = True
- self._recv_loop = _RecvLoop(self,
+ self._recv_loop = _RecvLoop(conn=self,
q_to_main=self._q_to_main,
bonus_iterator=self._read_lines())
- self.broadcast_conn(_ConnectedEvent)
+ self.broadcast(_ConnectedEvent)
Thread(target=connect, daemon=True, args=(self,)).start()
+ def broadcast[E: Event](self,
+ event_class: type[E],
+ *args, **kwargs
+ ) -> None:
+ 'Broadcast event subclassing ConnEvent, with self as its .conn.'
+ super().broadcast(event_class, conn=self, *args, **kwargs)
+
+ def send(self, verb: str, parameters: tuple[str, ...]) -> None:
+ 'Broadcast _SendEvent for _IrcMessage(verb, parameters).'
+ self.broadcast(_SendEvent, _IrcMessage(verb, parameters))
+
def update_login(self, **kwargs) -> None:
- 'Adapt ._login attributes to kwargs, broadcast NickSetEvent.'
+ 'Adapt .login attributes to kwargs, broadcast NickSetEvent.'
for key, val in kwargs.items():
- setattr(self._login, key, val)
- self.broadcast_conn(NickSetEvent)
+ setattr(self.login, key, val)
+ self.broadcast(NickSetEvent)
def close(self) -> None:
'Close both RecvLoop and socket.'
def _write_line(self, line: str) -> None:
'Send line-separator-delimited message over socket.'
if not (self._socket and self._assumed_open):
- self.broadcast_conn(LogConnEvent,
- 'ALERT: cannot send, assuming connection '
- 'closed')
+ self.broadcast(LogConnEvent,
+ 'ALERT: cannot send, assuming connection closed.')
return
self._socket.sendall(line.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR)
'Process connection-directed Event into further steps.'
if isinstance(event, InitReconnectEvent):
if self._assumed_open:
- self.broadcast_conn(LogConnEvent,
- 'ALERT: Reconnect called, but still seem '
- 'connected, so nothing to do.')
+ self.broadcast(LogConnEvent,
+ 'ALERT: Reconnect called, but still seem '
+ 'connected, so nothing to do.')
else:
self._start_connecting()
elif isinstance(event, _ConnectedEvent):
# self.send('CAP', ('LS', '302'))
- self.send('USER', (self._login.user, '0', '*', self._login.real))
- self.send('NICK', (self._login.nick,))
+ self.send('USER', (self.login.user, '0', '*', self.login.real))
+ self.send('NICK', (self.login.nick,))
# self.send('CAP', ('LIST',))
# self.send('CAP', ('END',))
elif isinstance(event, _DisconnectedEvent):
self.close()
elif isinstance(event, _SendEvent):
- self.broadcast_conn(LogConnEvent, f'->: {event.payload.raw}')
+ self.broadcast(LogConnEvent, f'->: {event.payload.raw}')
self._write_line(event.payload.raw)
-class _RecvLoop(Loop):
+class _RecvLoop(Loop, ConnMixin):
'Loop to react on messages from server.'
- def __init__(self, conn: IrcConnection, **kwargs) -> None:
- super().__init__(**kwargs)
- self._conn = conn
-
def process_bonus(self, yielded: str) -> None:
msg = _IrcMessage.from_raw(yielded)
- self._conn.broadcast_conn(LogConnEvent, f'<-: {msg.raw}')
+ self.conn.broadcast(LogConnEvent, f'<-: {msg.raw}')
if msg.verb == 'PING':
- self._conn.send('PONG', (msg.parameters[0],))
+ self.conn.send('PONG', (msg.parameters[0],))
elif msg.verb == 'ERROR'\
and msg.parameters[0].startswith('Closing link:'):
- self._conn.broadcast_conn(_DisconnectedEvent)
+ self.conn.broadcast(_DisconnectedEvent)
elif msg.verb in {'001', 'NICK'}:
- self._conn.update_login(nick=msg.parameters[0],
- nick_confirmed=True)
+ self.conn.update_login(nick=msg.parameters[0],
+ nick_confirmed=True)
from ircplom.events import (BroadcastMixin, Event, EventQueue, Loop,
PayloadMixin, QuitEvent)
from ircplom.irc_conn import (
- BroadcastConnMixin, ConnEvent, InitConnectEvent, InitConnWindowEvent,
- InitReconnectEvent, LoginNames, LogConnEvent, NickSetEvent,
- TIMEOUT_LOOP)
+ ConnEvent, ConnMixin, InitConnectEvent, InitReconnectEvent,
+ LoginNames, LogConnEvent, NickSetEvent, TIMEOUT_LOOP)
_MIN_HEIGHT = 4
_MIN_WIDTH = 32
'Defines most basic TUI object API.'
@abstractmethod
- def __init__(self, *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
+ def __init__(self, **kwargs) -> None:
+ super().__init__(**kwargs)
self.tainted = True
self._drawable = False
'Defines some API shared between _PromptWidget and _LogWidget.'
_history_idx: int
- def __init__(self, write: Callable[..., None], *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
+ def __init__(self, write: Callable[..., None], **kwargs) -> None:
+ super().__init__(**kwargs)
self._write = write
self._history: list[str] = []
_view_size: _YX
_y_pgscroll: int
- def __init__(self, wrap: Callable[[str], list[str]], *args, **kwargs
+ def __init__(self, wrap: Callable[[str], list[str]], **kwargs
) -> None:
- super().__init__(*args, **kwargs)
+ super().__init__(**kwargs)
self._wrap = wrap
self._wrapped_idx = self._history_idx = -1
self._wrapped: list[tuple[Optional[int], str]] = []
_input_buffer_unsafe: str
_cursor_x: int
- def __init__(self, *args, **kwargs) -> None:
- super().__init__(*args, **kwargs)
+ def __init__(self, **kwargs) -> None:
+ super().__init__(**kwargs)
self._reset_buffer('')
@property
return to_return
-class _ConnectionPromptWidget(_PromptWidget):
+class _ConnectionPromptWidget(_PromptWidget, ConnMixin):
'PromptWidget with attributes, methods for dealing with an IrcConnection.'
- login: LoginNames
@property
def _prompt(self) -> str:
- return ((' ' if self.login.nick_confirmed else '?')
- + self.login.nick
+ return ((' ' if self.conn.login.nick_confirmed else '?')
+ + self.conn.login.nick
+ super()._prompt)
super().__init__(**kwargs)
self.idx = idx
self._term = term
- self.log = _LogWidget(self._term.wrap, self._term.write)
- self.prompt = self.__annotations__['prompt'](self._term.write)
+ self.log = _LogWidget(wrap=self._term.wrap, write=self._term.write)
+ self.prompt = self.__annotations__['prompt'](write=self._term.write,
+ **kwargs)
if hasattr(self._term, 'size'):
self.set_geometry()
widget.draw()
-class _ConnectionWindow(_Window, BroadcastConnMixin):
+class _ConnectionWindow(_Window, ConnMixin):
'Window with attributes and methods for dealing with an IrcConnection.'
prompt: _ConnectionPromptWidget
- def __init__(self, login: LoginNames, **kwargs) -> None:
- super().__init__(**kwargs)
- self.prompt.login = login
-
def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None:
'Send QUIT command to server.'
- self.send('QUIT', (quit_msg, ))
+ self.conn.send('QUIT', (quit_msg, ))
def cmd__reconnect(self) -> None:
'Attempt reconnection.'
- self.broadcast_conn(InitReconnectEvent)
+ self.conn.broadcast(InitReconnectEvent)
def cmd__nick(self, new_nick: str) -> None:
'Attempt nickname change.'
- self.send('NICK', (new_nick, ))
+ self.conn.send('NICK', (new_nick, ))
class _KeyboardLoop(Loop, BroadcastMixin):
cmd = self._cmd_name_to_cmd(event.payload[0])
assert cmd is not None
cmd(*event.payload[1:])
- elif isinstance(event, InitConnWindowEvent):
- conn_win = _ConnectionWindow(q_to_main=self._q_to_main,
- conn_idx=event.conn_idx,
- idx=len(self._windows),
- term=self._term,
- login=event.payload)
- self._windows += [conn_win]
- self._conn_windows += [conn_win]
- self._switch_window(conn_win.idx)
elif isinstance(event, ConnEvent):
- conn_win = self._conn_windows[event.conn_idx]
+ matching_wins = [cw for cw in self._conn_windows
+ if cw.conn == event.conn]
+ if matching_wins:
+ conn_win = matching_wins[0]
+ else:
+ conn_win = _ConnectionWindow(idx=len(self._windows),
+ conn=event.conn,
+ term=self._term)
+ self._windows += [conn_win]
+ self._conn_windows += [conn_win]
+ self._switch_window(conn_win.idx)
if isinstance(event, LogConnEvent):
conn_win.log.append(event.payload)
elif isinstance(event, NickSetEvent):
projects / ircplom / commitdiff