(r'\\', '\\'))
+class _IrcMessage:
+ 'Properly structured representation of IRC message as per IRCv3 spec.'
+ _raw: Optional[str] = None
+
+ def __init__(self,
+ verb: str,
+ parameters: Optional[tuple[str, ...]] = None,
+ source: str = '',
+ tags: Optional[dict[str, str]] = None
+ ) -> None:
+ self.verb: str = verb
+ self.parameters: tuple[str, ...] = parameters or tuple()
+ self.source: str = source
+ self.tags: dict[str, str] = tags or {}
+
+ @classmethod
+ def from_raw(cls, raw_msg: str) -> Self:
+ 'Parse raw IRC message line into properly structured _IrcMessage.'
+
+ class _Stage(NamedTuple):
+ name: str
+ prefix_char: Optional[str]
+ processor: Callable = lambda s: s
+
+ def _parse_tags(str_tags: str) -> dict[str, str]:
+ tags = {}
+ for str_tag in [s for s in str_tags.split(';') if s]:
+ if '=' in str_tag:
+ key, val = str_tag.split('=', maxsplit=1)
+ for to_repl, repl_with in _IRCSPEC_TAG_ESCAPES:
+ val = val.replace(to_repl, repl_with)
+ else:
+ key, val = str_tag, ''
+ tags[key] = val
+ return tags
+
+ def _split_params(str_params: str) -> tuple[str, ...]:
+ params = []
+ params_stage = 0 # 0: gap, 1: non-trailing, 2: trailing
+ for char in str_params:
+ if char == ' ' and params_stage < 2:
+ params_stage = 0
+ continue
+ if params_stage == 0:
+ params += ['']
+ params_stage += 1
+ if char == ':':
+ params_stage += 1
+ continue
+ params[-1] += char
+ return tuple(p for p in params)
+
+ stages = [_Stage('tags', '@', _parse_tags),
+ _Stage('source', ':'),
+ _Stage('verb', None, lambda s: s.upper()),
+ _Stage('parameters', None, _split_params)]
+ harvest = {s.name: '' for s in stages}
+ idx_stage = 0
+ stage = None
+ for char in raw_msg:
+ if char == ' ' and idx_stage < (len(stages) - 1):
+ if stage:
+ stage = None
+ continue
+ if not stage:
+ while not stage:
+ idx_stage += 1
+ tested = stages[idx_stage]
+ if (not tested.prefix_char) or char == tested.prefix_char:
+ stage = tested
+ if stage.prefix_char:
+ continue
+ harvest[stage.name] += char
+ msg = cls(**{s.name: s.processor(harvest[s.name]) for s in stages})
+ msg._raw = raw_msg
+ return msg
+
+ @property
+ def raw(self) -> str:
+ 'Return raw message code – create from known fields if necessary.'
+ if not self._raw:
+ to_combine = []
+ if self.tags:
+ tag_strs = []
+ for key, val in self.tags.items():
+ tag_strs += [key]
+ if not val:
+ continue
+ for repl_with, to_repl in reversed(_IRCSPEC_TAG_ESCAPES):
+ val = val.replace(to_repl, repl_with)
+ tag_strs[-1] += f'={val}'
+ to_combine += ['@' + ';'.join(tag_strs)]
+ to_combine += [self.verb]
+ if self.parameters:
+ to_combine += self.parameters[:-1]
+ to_combine += [f':{self.parameters[-1]}']
+ self._raw = ' '.join(to_combine)
+ return self._raw
+
+
class _ConnIdxMixin:
'Collects a Connection ID at .conn_idx.'
self._login = login
self._socket: Optional[socket] = None
self._assumed_open = False
- self._loop: Optional[_ConnectionLoop] = None
+ self._recv_loop: Optional[_RecvLoop] = None
self.broadcast_conn(InitConnWindowEvent, self._login)
self._start_connecting()
return
self._socket.settimeout(TIMEOUT_LOOP)
self._assumed_open = True
- self._loop = _ConnectionLoop(self,
- q_to_main=self._q_to_main,
- bonus_iterator=self._read_lines())
+ self._recv_loop = _RecvLoop(self,
+ q_to_main=self._q_to_main,
+ bonus_iterator=self._read_lines())
self.broadcast_conn(_ConnectedEvent)
Thread(target=connect, daemon=True, args=(self,)).start()
self.broadcast_conn(NickSetEvent)
def close(self) -> None:
- 'Close both _ConnectionLoop and socket.'
+ 'Close both RecvLoop and socket.'
self._assumed_open = False
self.update_login(nick_confirmed=False)
- if self._loop:
- self._loop.stop()
- self._loop = None
+ if self._recv_loop:
+ self._recv_loop.stop()
+ self._recv_loop = None
if self._socket:
self._socket.close()
self._socket = None
self._write_line(event.payload.raw)
-class _IrcMessage:
- 'Properly structured representation of IRC message as per IRCv3 spec.'
- _raw: Optional[str] = None
-
- def __init__(self,
- verb: str,
- parameters: Optional[tuple[str, ...]] = None,
- source: str = '',
- tags: Optional[dict[str, str]] = None
- ) -> None:
- self.verb: str = verb
- self.parameters: tuple[str, ...] = parameters or tuple()
- self.source: str = source
- self.tags: dict[str, str] = tags or {}
-
- @classmethod
- def from_raw(cls, raw_msg: str) -> Self:
- 'Parse raw IRC message line into properly structured _IrcMessage.'
-
- class _Stage(NamedTuple):
- name: str
- prefix_char: Optional[str]
- processor: Callable = lambda s: s
-
- def _parse_tags(str_tags: str) -> dict[str, str]:
- tags = {}
- for str_tag in [s for s in str_tags.split(';') if s]:
- if '=' in str_tag:
- key, val = str_tag.split('=', maxsplit=1)
- for to_repl, repl_with in _IRCSPEC_TAG_ESCAPES:
- val = val.replace(to_repl, repl_with)
- else:
- key, val = str_tag, ''
- tags[key] = val
- return tags
-
- def _split_params(str_params: str) -> tuple[str, ...]:
- params = []
- params_stage = 0 # 0: gap, 1: non-trailing, 2: trailing
- for char in str_params:
- if char == ' ' and params_stage < 2:
- params_stage = 0
- continue
- if params_stage == 0:
- params += ['']
- params_stage += 1
- if char == ':':
- params_stage += 1
- continue
- params[-1] += char
- return tuple(p for p in params)
-
- stages = [_Stage('tags', '@', _parse_tags),
- _Stage('source', ':'),
- _Stage('verb', None, lambda s: s.upper()),
- _Stage('parameters', None, _split_params)]
- harvest = {s.name: '' for s in stages}
- idx_stage = 0
- stage = None
- for char in raw_msg:
- if char == ' ' and idx_stage < (len(stages) - 1):
- if stage:
- stage = None
- continue
- if not stage:
- while not stage:
- idx_stage += 1
- tested = stages[idx_stage]
- if (not tested.prefix_char) or char == tested.prefix_char:
- stage = tested
- if stage.prefix_char:
- continue
- harvest[stage.name] += char
- msg = cls(**{s.name: s.processor(harvest[s.name]) for s in stages})
- msg._raw = raw_msg
- return msg
-
- @property
- def raw(self) -> str:
- 'Return raw message code – create from known fields if necessary.'
- if not self._raw:
- to_combine = []
- if self.tags:
- tag_strs = []
- for key, val in self.tags.items():
- tag_strs += [key]
- if not val:
- continue
- for repl_with, to_repl in reversed(_IRCSPEC_TAG_ESCAPES):
- val = val.replace(to_repl, repl_with)
- tag_strs[-1] += f'={val}'
- to_combine += ['@' + ';'.join(tag_strs)]
- to_combine += [self.verb]
- if self.parameters:
- to_combine += self.parameters[:-1]
- to_combine += [f':{self.parameters[-1]}']
- self._raw = ' '.join(to_combine)
- return self._raw
-
-
-class _ConnectionLoop(Loop):
+class _RecvLoop(Loop):
'Loop to react on messages from server.'
def __init__(self, conn: IrcConnection, **kwargs) -> None:
projects / ircplom / commitdiff