'IRC server connection management.'
# built-ins
+from dataclasses import dataclass
from socket import socket, gaierror as socket_gaierror
from threading import Thread
from typing import Callable, Iterator, NamedTuple, Optional, Self
(r'\\', '\\'))
-class LoginNames(NamedTuple):
+@dataclass
+class LoginNames:
'Collects the names needed on server connect for USER, NICK commands.'
user: str
nick: str
real: str
+ nick_confirmed: bool = False
class InitConnectEvent(Event, PayloadMixin):
class InitConnWindowEvent(ConnEvent, PayloadMixin):
- 'Event to trigger TUI making ConnectionWindow, nick in prompt = payload.'
- payload: str
+ 'Event to trigger TUI making ConnectionWindow.'
+ payload: LoginNames
-class _ConnectedEvent(ConnEvent, PayloadMixin):
- 'Event to signal opening of connection, with payload login names to send.'
- payload: LoginNames
+class _ConnectedEvent(ConnEvent):
+ 'Event to signal opening of connection.'
-class DisconnectedEvent(ConnEvent):
+class _DisconnectedEvent(ConnEvent):
'Event to signal closing of connection'
payload: str
-class NickSetEvent(ConnEvent, PayloadMixin):
- 'Event to signal nickname (= payload) having been set server-side.'
- payload: str
+class NickSetEvent(ConnEvent):
+ 'Event to signal nickname having been set server-side.'
class SendEvent(ConnEvent, PayloadMixin):
self._socket: Optional[socket] = None
self._assumed_open = False
self._loop: Optional[_ConnectionLoop] = None
- self.broadcast_conn(InitConnWindowEvent, self._login.nick)
+ self.broadcast_conn(InitConnWindowEvent, self._login)
self._start_connecting()
def _start_connecting(self) -> None:
self._assumed_open = True
self._loop = _ConnectionLoop(conn_idx=self.conn_idx,
q_to_main=self._q_to_main,
- bonus_iterator=self._read_lines())
- self.broadcast_conn(_ConnectedEvent, self._login)
+ bonus_iterator=self._read_lines(),
+ login=self._login,
+ set_login=self._set_login)
+ self.broadcast_conn(_ConnectedEvent)
Thread(target=connect, daemon=True, args=(self,)).start()
+ def _set_login(self, **kwargs) -> None:
+ for key, val in kwargs.items():
+ setattr(self._login, key, val)
+ self.broadcast_conn(NickSetEvent)
+
def close(self) -> None:
'Close both _ConnectionLoop and socket.'
self._assumed_open = False
+ self._set_login(nick_confirmed=False)
if self._loop:
self._loop.stop()
self._loop = None
elif isinstance(event, _ConnectedEvent):
assert self._loop is not None
self._loop.put(event)
- elif isinstance(event, DisconnectedEvent):
+ elif isinstance(event, _DisconnectedEvent):
self.close()
elif isinstance(event, SendEvent):
self.broadcast_conn(LogConnEvent, f'->: {event.payload.raw}')
class _ConnectionLoop(Loop, BroadcastConnMixin):
'Loop receiving and translating socket messages towards main loop.'
+ def __init__(self, login: LoginNames, set_login: Callable, **kwargs
+ ) -> None:
+ super().__init__(**kwargs)
+ self._login = login
+ self._set_login = set_login
+
def _send(self, verb: str, parameters: tuple[str, ...]) -> None:
self.broadcast_conn(SendEvent, IrcMessage(verb, parameters))
return False
if isinstance(event, _ConnectedEvent):
# self._send('CAP', ('LS', '302'))
- self._send('USER', (event.payload.user, '0', '*',
- event.payload.real))
- self._send('NICK', (event.payload.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',))
return True
self._send('PONG', (msg.parameters[0],))
elif msg.verb == 'ERROR'\
and msg.parameters[0].startswith('Closing link:'):
- self.broadcast_conn(DisconnectedEvent)
+ self.broadcast_conn(_DisconnectedEvent)
elif msg.verb in {'001', 'NICK'}:
- self.broadcast_conn(NickSetEvent, msg.parameters[0])
+ self._set_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, DisconnectedEvent, InitConnectEvent,
- InitConnWindowEvent, InitReconnectEvent, IrcMessage, LoginNames,
- LogConnEvent, NickSetEvent, SendEvent, TIMEOUT_LOOP)
+ BroadcastConnMixin, ConnEvent, InitConnectEvent, InitConnWindowEvent,
+ InitReconnectEvent, IrcMessage, LoginNames, LogConnEvent, NickSetEvent,
+ SendEvent, TIMEOUT_LOOP)
_MIN_HEIGHT = 4
_MIN_WIDTH = 32
class _ConnectionPromptWidget(_PromptWidget):
'PromptWidget with attributes, methods for dealing with an IrcConnection.'
- _nick: str = ''
- _nick_confirmed: bool = False
+ login: LoginNames
@property
def _prompt(self) -> str:
- return ((' ' if self._nick_confirmed else '?')
- + self._nick
+ return ((' ' if self.login.nick_confirmed else '?')
+ + self.login.nick
+ super()._prompt)
- def update_prompt(self,
- nick_confirmed=False,
- nick: Optional[str] = None
- ) -> None:
- 'Update nickname-relevant knowledge to go into prompt string.'
- self.tainted = True
- self._nick_confirmed = nick_confirmed
- if nick:
- self._nick = nick
-
class _Window(_Widget):
'Collects a log and a prompt meant for the same content stream.'
'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.broadcast_conn(SendEvent, IrcMessage('QUIT', (quit_msg, )))
conn_win = _ConnectionWindow(q_to_main=self._q_to_main,
conn_idx=event.conn_idx,
idx=len(self._windows),
- term=self._term)
- conn_win.prompt.update_prompt(nick_confirmed=False,
- nick=event.payload)
+ term=self._term,
+ login=event.payload)
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):
- conn_win.prompt.update_prompt(nick_confirmed=True,
- nick=event.payload)
- elif isinstance(event, DisconnectedEvent):
- conn_win.prompt.update_prompt(nick_confirmed=False)
+ conn_win.prompt.tainted = True
else:
return True
self.window.draw_tainted()
projects / ircplom / commitdiff