if prefix:
self._add_membership_prefix(user_id, prefix)
- def add_user(self, user: '_User') -> None:
- 'To .user_ids add user.nickname, keep .user_ids declared complete.'
- user_id = self._userid_for_nickuserhost(user, create_if_none=True,
+ def join_user(self, user: '_User') -> None:
+ 'Register non-"me" user joining channel.'
+ user_id = self._userid_for_nickuserhost(nickuserhost=user,
+ create_if_none=True,
updating=True)
- self.user_ids.completable_add(user_id, on_complete=True)
+ if user_id != 'me': # own JOIN would have added name via RPL_NAMREPLY
+ self.user_ids.completable_add(user_id, on_complete=True)
def remove_user(self, user: '_User', msg: str) -> None:
'From .user_ids remove .nickname, keep .user_ids declared complete.'
attr._preset_init_kwargs = {}
return attr
+ def set_isupport_from_rpl(self, rpl: tuple[str, ...]) -> None:
+ 'Parse rpl for additions/deletions to .isupport dict.'
+ for item in rpl:
+ if item[0] == '-':
+ del self.isupport[item[1:]]
+ else:
+ key, data = _tuple_key_val_from_eq_str(item)
+ self.isupport[key] = data
+
def messaging(self, src: str | NickUserHost) -> ChatMessage:
'Start input chain for chat message data.'
return _ChatMessage(sender=src, db=self)
self.conn.send(msg)
return msg
+ def autojoin(self) -> None:
+ 'Join all channels marked for auto-join, clear same auto-join list.'
+ for name in self._autojoins:
+ self.send('JOIN', name)
+ self._autojoins.clear()
+
+ def send_authentication(self) -> None:
+ 'Send AUTHENTICATE with b64-encoded authentication data.'
+ auth = b64encode((self.db.nick_wanted + '\0'
+ + self.db.nick_wanted + '\0'
+ + self.db.password
+ ).encode('utf-8')
+ ).decode('utf-8')
+ self.send('AUTHENTICATE', auth)
+
+ def increment_for_nicknameinuse(self, rejected_nick: str) -> None:
+ 'Note ERR_NICKNAMEINUSE and try incremented variant of rejected_nick.'
+ self._alert('nickname already in use, trying increment')
+ self.send('NICK', _NickUserHost(nick=rejected_nick).incremented)
+
+ def check_pong(self, reply: str) -> None:
+ 'Check reply matching our PING, zero our PING expectations.'
+ assert self._expected_pong == reply
+ self._expected_pong = ''
+
+ def pong(self, reply: str) -> None:
+ 'Reply PING with its expected reply.'
+ self.send('PONG', reply)
+
def handle_msg(self, msg: IrcMessage) -> None:
'Log msg.raw, then process incoming msg into appropriate client steps.'
ret = {}
node(ret[tok_name])
else:
node()
- if ret['_verb'] == '001': # RPL_WELCOME
- for name in self._autojoins:
- self.send('JOIN', name)
- self._autojoins.clear()
- elif ret['_verb'] == '005': # RPL_ISUPPORT
- for item in ret['isupport']:
- if item[0] == '-':
- del self.db.isupport[item[1:]]
- else:
- key, data = _tuple_key_val_from_eq_str(item)
- self.db.isupport[key] = data
- elif ret['_verb'] == '401': # ERR_NOSUCHNICK
+ if ret['_verb'] == '401': # ERR_NOSUCHNICK
raise TargetUserOffline(ret['missing'])
- elif ret['_verb'] == '432': # ERR_ERRONEOUSNICKNAME
+ if ret['_verb'] == '432': # ERR_ERRONEOUSNICKNAME
self._alert('nickname refused for bad format, '
+ ('keeping current one' if 'nick' in ret
else 'giving up'))
- elif ret['_verb'] == '433': # ERR_NICKNAMEINUSE
- self._alert('nickname already in use, trying increment')
- self.send('NICK', _NickUserHost(nick=ret['used']).incremented)
- elif ret['_verb'] == 'AUTHENTICATE':
- auth = b64encode((self.db.nick_wanted + '\0'
- + self.db.nick_wanted + '\0'
- + self.db.password
- ).encode('utf-8')).decode('utf-8')
- self.send('AUTHENTICATE', auth)
elif ret['_verb'] == 'CAP':
if (self.caps.process_msg(verb=ret['subverb'],
items=ret['items'],
self.send('AUTHENTICATE', _SASL_PLAIN)
else:
self.caps.end_negotiation()
- elif ret['_verb'] == 'JOIN' and ret['joiner'] != self.db.users['me']:
- self.db.channels[ret['channel']].add_user(ret['joiner'])
elif ret['_verb'] == 'MODE' and 'mode_on_nick' in ret:
self.db.channels[ret['channel']].mode_on_nick(
ret['nick'], ret['mode_on_nick'])
if ret['parter'] is self.db.users['me']:
del self.db.channels[ret['channel']]
self.db.users.purge()
- elif ret['_verb'] == 'PING':
- self.send('PONG', ret['reply'])
- elif ret['_verb'] == 'PONG':
- assert self._expected_pong == ret['reply']
- self._expected_pong = ''
ClientsDb = dict[str, Client]
'001', # RPL_WELCOME
_MsgTok.SERVER,
((_MsgTok.NICKNAME, 'setattr_db.users.me:nick'),
- _MsgTok.ANY)),
+ _MsgTok.ANY),
+ bonus_tasks=('do_autojoin:',)),
_MsgParseExpectation(
'002', # RPL_YOURHOST
'005', # RPL_ISUPPORT
_MsgTok.SERVER,
((_MsgTok.NICKNAME, 'setattr_db.users.me:nick'),
- (_MsgTok.ANY, ':isupport'),
+ (_MsgTok.ANY, 'do_db.set_isupport_from_rpl:isupport'),
_MsgTok.ANY), # comment
idx_into_list=1),
_MsgParseExpectation(
'AUTHENTICATE',
_MsgTok.NONE,
- ('+',)),
+ ('+',),
+ bonus_tasks=('do_send_authentication:',)),
# capability negotation
'433', # ERR_NICKNAMEINUSE
_MsgTok.SERVER,
('*',
- (_MsgTok.NICKNAME, ':used'),
+ (_MsgTok.NICKNAME, 'do_increment_for_nicknameinuse:rejected'),
_MsgTok.ANY)), # comment
_MsgParseExpectation(
'433', # ERR_NICKNAMEINUSE
_MsgTok.SERVER,
(_MsgTok.NICKNAME, # we rather go for incrementation
- (_MsgTok.NICKNAME, ':used'),
+ (_MsgTok.NICKNAME, 'do_increment_for_nicknameinuse:rejected'),
_MsgTok.ANY)), # comment
_MsgParseExpectation(
_MsgParseExpectation(
'JOIN',
- (_MsgTok.NICK_USER_HOST, ':joiner'),
- ((_MsgTok.CHANNEL, ':channel'),)),
+ (_MsgTok.NICK_USER_HOST, 'do_db.channels.CHANNEL.join_user:user'),
+ ((_MsgTok.CHANNEL, ':CHANNEL'),)),
_MsgParseExpectation(
'PART',
_MsgParseExpectation(
'PING',
_MsgTok.NONE,
- ((_MsgTok.ANY, ':reply'),)),
+ ((_MsgTok.ANY, 'do_pong:reply'),)),
_MsgParseExpectation(
'PONG',
_MsgTok.SERVER,
(_MsgTok.SERVER,
- (_MsgTok.ANY, ':reply'),)),
+ (_MsgTok.ANY, 'do_check_pong:reply'),)),
# misc.
projects / ircplom / commitdiff