AffectiveEvent, ExceptionEvent, Loop, PayloadMixin, QueueMixin)
+ClientsDb = dict[UUID, 'Client']
+
+CHAT_GLOB = '*'
+
_TIMEOUT_RECV_LOOP = 0.1
_TIMEOUT_CONNECT = 5
_CONN_RECV_BUFSIZE = 1024
(r'\r', '\r'),
(r'\\', '\\'))
-ClientsDb = dict[UUID, 'Client']
-
class IrcMessage:
'Properly structured representation of IRC message as per IRCv3 spec.'
class _ConnectedEvent(ClientEvent):
def affect(self, target: 'Client') -> None:
+ target.log(msg='# connected to server', chat=CHAT_GLOB)
target.send(IrcMessage(verb='USER', params=(getuser(), '0', '*',
target.realname)))
target.send(IrcMessage(verb='NICK', params=(target.nickname,)))
def affect(self, target: 'Client') -> None:
if target.assumed_open:
- target.log('ALERT: Reconnect called, but still seem connected, '
- 'so nothing to do.')
+ target.log('# ALERT: reconnection called, but still seem '
+ 'connected, so nothing to do.')
else:
target.start_connecting()
'To trigger sending of payload to server.'
payload: IrcMessage
- def affect(self, target: 'Client') -> None:
- target.send(self.payload)
+ def affect(self, target: 'Client', chat: str = '') -> None:
+ target.send(msg=self.payload, chat=chat)
class ClientQueueMixin(QueueMixin):
class Client(ABC, ClientQueueMixin):
'Abstracts socket connection, loop over it, and handling messages from it.'
+ nick_confirmed: bool
+ nickname: str
def __init__(self, hostname: str, nickname: str, realname: str, **kwargs
) -> None:
super().__init__(**kwargs)
- self.id_ = uuid4()
self._hostname = hostname
self._socket: Optional[socket] = None
- self.assumed_open = False
self._recv_loop: Optional[Loop] = None
+ self.id_ = uuid4()
+ self.assumed_open = False
self.realname = realname
self.update_login(nick_confirmed=False, nickname=nickname)
self.start_connecting()
def connect(self) -> None:
try:
self._socket = socket()
- self.log(f'Connecting to {self._hostname} …')
+ self.log(f'# connecting to server {self._hostname} …')
self._socket.settimeout(_TIMEOUT_CONNECT)
try:
self._socket.connect((self._hostname, _PORT))
except (TimeoutError, socket_gaierror) as e:
- self.log(f'ALERT: {e}')
+ self.log(f'# ALERT: {e}')
return
self._socket.settimeout(_TIMEOUT_RECV_LOOP)
self.assumed_open = True
@abstractmethod
def log(self, msg: str, chat: str = '') -> None:
- 'Write msg into log, whatever shape that may have.'
+ 'Write msg into log of chat, whatever shape that may have.
+
+ Messages to chat=CHAT_GLOB are meant to appear in all widgets mapped to
+ the client, those to chat="" only in the initial connection window.
+ '
- def send(self, msg: IrcMessage) -> None:
+ def send(self, msg: IrcMessage, chat: str = '') -> None:
'Send line-separator-delimited message over socket.'
if not (self._socket and self.assumed_open):
- self.log('ALERT: cannot send, assuming connection closed.')
+ self.log('# ALERT: cannot send, connection seems closed')
return
self._socket.sendall(msg.raw.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR)
- self.log(f'->: {msg.raw}')
+ self.log(msg=f'->: {msg.raw}', chat=chat)
def update_login(self, nick_confirmed: bool, nickname: str = '') -> None:
'Manage .nickname, .nick_confirmed – useful for subclass extension.'
- if nickname:
+ first_run = not hasattr(self, 'nickname')
+ prefix = '# nickname'
+ if first_run or (nickname and nickname != self.nickname):
+ verb = 'set' if first_run else f'changed from "{self.nickname}'
self.nickname = nickname
- self.nick_confirmed = nick_confirmed
+ self.log(msg=f'{prefix} {verb} to {nickname}', chat=CHAT_GLOB)
+ if first_run or nick_confirmed != self.nick_confirmed:
+ self.nick_confirmed = nick_confirmed
+ if not first_run:
+ self.log(
+ msg=f'{prefix} {"" if nick_confirmed else "un"}confirmed')
def close(self) -> None:
'Close both recv Loop and socket.'
+ self.log(msg='# disconnected from server', chat=CHAT_GLOB)
self.assumed_open = False
self.update_login(nick_confirmed=False)
if self._recv_loop:
from ircplom.events import (
AffectiveEvent, Loop, PayloadMixin, QueueMixin, QuitEvent)
from ircplom.irc_conn import (
- IrcMessage, Client, ClientIdMixin, ClientQueueMixin,
+ CHAT_GLOB, IrcMessage, Client, ClientIdMixin, ClientQueueMixin,
InitReconnectEvent, NewClientEvent, SendEvent)
_MIN_HEIGHT = 4
elif len(self.payload) == 1:
target.window.prompt.insert(self.payload)
else:
- target.log(f'ALERT: unknown keyboard input: {self.payload}')
+ target.log(f'# ALERT: unknown keyboard input: {self.payload}')
super().affect(target)
'Currently selected _Window.'
return self.windows[self._window_idx]
+ 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,
+ client_id=client_id, chat=chat)
+ self.windows += [win]
+ self._switch_window(new_idx)
+ return win
+
def client_wins(self, client_id: UUID) -> list['_ClientWindow']:
- 'All _ClientWindows matching client_id.'
- return [win for win in self.windows
+ 'All _ClientWindows matching client_id; if none, create one.'
+ wins = [win for win in self.windows
if isinstance(win, _ClientWindow)
and win.client_id == client_id] # pylint: disable=no-member
+ if not wins:
+ wins = [self._new_client_window(client_id=client_id)]
+ return wins
def client_win(self, client_id: UUID, chat: str = '') -> '_ClientWindow':
'''That _ClientWindow matching client_id and chat; create if none.
- In case of creation, also switch to window, and if not client's first
- window, copy prompt prefix from client's first window.
+ In case of creation, copy prompt prefix from client's first window.
'''
client_wins = self.client_wins(client_id)
candidates = [win for win in client_wins if win.chat == chat]
if candidates:
return candidates[0]
- new_idx = len(self.windows)
- win = _ClientWindow(idx=new_idx, term=self.term, q_out=self._q_out,
- client_id=client_id, chat=chat)
+ win = self._new_client_window(client_id=client_id, chat=chat)
if client_wins:
win.prompt.prefix = client_wins[0].prompt.prefix
- self.windows += [win]
- self._switch_window(new_idx)
return win
def log(self, msg: str) -> None:
else:
alert = 'not prefixed by /'
if alert:
- self.log(f'invalid prompt command: {alert}')
+ self.log(f'# ALERT: invalid prompt command: {alert}')
def cmd__quit(self) -> None:
'Trigger program exit.'
def cmd__disconnect(self, quit_msg: str = 'ircplom says bye') -> None:
'Send QUIT command to server.'
- self._cput(SendEvent, payload=IrcMessage(verb='QUIT',
- params=(quit_msg,)))
+ self._cput(SendEvent,
+ payload=IrcMessage(verb='QUIT', params=(quit_msg,)))
def cmd__reconnect(self) -> None:
'Attempt reconnection.'
def cmd__nick(self, new_nick: str) -> None:
'Attempt nickname change.'
- self._cput(SendEvent, payload=IrcMessage(verb='NICK',
- params=(new_nick,)))
+ self._cput(SendEvent,
+ payload=IrcMessage(verb='NICK', params=(new_nick,)))
class _ClientWindowEvent(TuiEvent, ClientIdMixin):
payload: str
def affect(self, target: Tui) -> None:
- target.client_win(self.client_id, self.chat).log.append(self.payload)
+ if self.chat == CHAT_GLOB:
+ for win in target.client_wins(self.client_id):
+ win.log.append(self.payload)
+ else:
+ target.client_win(self.client_id, self.chat
+ ).log.append(self.payload)
super().affect(target)
projects / ircplom / commitdiff