AffectiveEvent, CrashingException, ExceptionEvent, QueueMixin)
from ircplom.irc_conn import (
BaseIrcConnection, IrcConnException, IrcMessage, NickUserHost,
- ERR_STR_TIMEOUT, ILLEGAL_NICK_CHARS, ILLEGAL_NICK_FIRSTCHARS,
+ SendFail, ERR_STR_TIMEOUT, ILLEGAL_NICK_CHARS, ILLEGAL_NICK_FIRSTCHARS,
ISUPPORT_DEFAULTS, PORT_SSL)
from ircplom.msg_parse_expectations import MSG_EXPECTATIONS
return toks[0], ('' if len(toks) == 1 else toks[1])
-class SendFail(Exception):
- 'When Client.send fails.'
-
-
class TargetUserOffline(Exception):
'When according to server our target user is not to be found.'
# ourselves
from ircplom.client import (
Channel, ChatMessage, Client, ClientQueueMixin, ImplementationFail,
- IrcConnSetup, NewClientEvent, SendFail, ServerCapability,
- SharedClientDbFields, TargetUserOffline, User)
+ IrcConnSetup, NewClientEvent, ServerCapability, SharedClientDbFields,
+ TargetUserOffline, User)
from ircplom.db_primitives import (
AutoAttrMixin, DbLinked, DbLinking, Dict, DictItem)
-from ircplom.irc_conn import IrcMessage, NickUserHost
+from ircplom.irc_conn import IrcMessage, NickUserHost, SendFail
from ircplom.tui_base import (
BaseTui, PromptWidget, StylingString, TuiEvent, Window,
CMD_SHORTCUTS, LOG_FMT_ATTRS, LOG_FMT_TAG_ALERT, LOG_PREFIX_DEFAULT)
(r'\n', '\n'),
(r'\r', '\r'),
(r'\\', '\\'))
+_LEN_MAX_MESSAGE = 512
+
+
+class SendFail(Exception):
+ 'Failure to send message.'
@dataclass
self._recv_loop.stop()
self._socket.close()
+ def _make_sendable(self, msg: IrcMessage) -> bytes:
+ to_send = msg.raw.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR
+ if len(to_send) > _LEN_MAX_MESSAGE:
+ raise SendFail('message too long')
+ return to_send
+
def send(self, msg: IrcMessage) -> None:
'Send line-separator-delimited message over socket.'
- self._socket.sendall(msg.raw.encode('utf-8') + _IRCSPEC_LINE_SEPARATOR)
+ self._socket.sendall(self._make_sendable(msg))
@abstractmethod
def _make_recv_event(self, msg: IrcMessage) -> Event:
self._recv_loop.stop()
def send(self, msg: IrcMessage) -> None:
- pass
+ self._make_sendable(msg)
def _read_lines(self) -> Iterator[Optional[Event]]:
try: