diff options
Diffstat (limited to 'packages/meshbay-cli/src/meshbay_cli/commands/chat.py')
| -rw-r--r-- | packages/meshbay-cli/src/meshbay_cli/commands/chat.py | 185 |
1 files changed, 168 insertions, 17 deletions
diff --git a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py index 6fa084d..d650af3 100644 --- a/packages/meshbay-cli/src/meshbay_cli/commands/chat.py +++ b/packages/meshbay-cli/src/meshbay_cli/commands/chat.py @@ -8,8 +8,17 @@ asks for another count. `chat GROUP -m TEXT` posts TEXT (`-m -` reads it from stdin), sealed and signed as the application seals and signs it. Each message is shown with its time and the name its sender sealed in it; one that cannot be read is shown as such, never left out. + +`chat GROUP --follow` prints the last messages, then each new one as the node +pushes it, until Ctrl+C. A connection that drops — or that stops answering the +ping sent every half minute, as a sleeping laptop's does — is opened again, +and what was said meanwhile is read from the history: each message is printed +once, whichever way it came (the node stores a device and nonce pair once). +With `--json`, one object per line. """ +import asyncio +import contextlib import json import re import sys @@ -18,7 +27,7 @@ from ..client.chat import chat_keys, history, open_row, post from ..client.groups import my_groups, open_group from ..client.paths import group_arg, handle, resolve_group from ..client.roster import RosterCheck -from ..errors import USAGE, CliError +from ..errors import NODE, USAGE, CliError from ..output import date, err, safe, sentence from .context import Context @@ -26,6 +35,10 @@ HELP = ("chat: alone, interactively; with a group, its last 10 messages (-N for "or post with -m") _COUNT = re.compile(r"^-(\d+)$") +# --follow: how often the connection is asked whether it is alive, how long it +# has to answer, and the waits between attempts to open it again. +PING_EVERY, PING_TIMEOUT = 30, 10 +RECONNECT_WAITS = (2, 5, 10, 30) # The one verdict worth a line (roster.py): the application's words. KEY_CHANGED = ("! This account is using a key you have not seen before: a new device " "added elsewhere, or the node naming the wrong person.") @@ -42,6 +55,9 @@ def add_parser(sub, parent) -> None: help="how many of the last messages to show (also -N, as -100)") p.add_argument("-m", "--message", metavar="TEXT", help="post TEXT instead of reading; - reads it from stdin") + p.add_argument("-f", "--follow", action="store_true", + help="after the last messages, print new ones as they arrive, until " + "Ctrl+C; with --json, one object per line") def rewrite_count(argv: list[str], actions) -> list[str]: @@ -69,8 +85,10 @@ async def run(ctx: Context) -> int: a = ctx.args if a.last < 1: raise CliError(USAGE, "Ask for one message at least.") + if a.follow and a.message is not None: + raise CliError(USAGE, "Post with -m, or follow with -f; not both at once.") if a.group is None: - if a.message is not None: + if a.message is not None or a.follow: raise CliError(USAGE, "Name the group to post to.") if not (sys.stdin.isatty() and sys.stdout.isatty()): raise CliError(USAGE, "The interactive chat needs a terminal; name a group " @@ -97,35 +115,168 @@ async def run(ctx: Context) -> int: check = RosterCheck(account, session) messages = [await open_row(r, keys, group["id"], check) for r in await history(session, a.last)] + if a.follow: + follower = Follower(ctx, account, group, session, keys, check) + for m in messages: + follower.emit(m) + if not messages and not a.quiet: + err(f"No message in {safe(handle(group))} yet; waiting for one " + "(Ctrl+C to stop).") + session = None # the follower owns it now + await follower.run() + return 0 finally: if session is not None: await session.close() await account.close() if a.json: - json.dump([{"id": m.id, "timestamp": m.timestamp, "sender": m.sender, - "text": m.text, "trust": m.trust, - **({"unreadable": m.unreadable} if m.unreadable else {})} - for m in messages], - sys.stdout, indent=1, ensure_ascii=False) + json.dump([as_json(m) for m in messages], sys.stdout, indent=1, ensure_ascii=False) print() return 0 if not messages and not a.quiet: err(f"No message in {safe(handle(group))} yet.") for m in messages: - head = f"{date(int(m.timestamp))} {safe(m.sender)}: " - if m.unreadable: - print(f"{head}[this message cannot be read: {m.unreadable}]") - continue - lines = safe_lines(m.text) - print(head + lines[0]) - for line in lines[1:]: - print(" " * len(head) + line) - if m.trust == "changed": - print(" " * len(head) + KEY_CHANGED) + print_message(m) return 0 +def print_message(m, *, flush: bool = False) -> None: + head = f"{date(int(m.timestamp))} {safe(m.sender)}: " + if m.unreadable: + print(f"{head}[this message cannot be read: {m.unreadable}]", flush=flush) + return + lines = safe_lines(m.text) + print(head + lines[0]) + for line in lines[1:]: + print(" " * len(head) + line) + if m.trust == "changed": + print(" " * len(head) + KEY_CHANGED) + if flush: + sys.stdout.flush() + + +def as_json(m) -> dict: + return {"id": m.id, "timestamp": m.timestamp, "sender": m.sender, "text": m.text, + "trust": m.trust, **({"unreadable": m.unreadable} if m.unreadable else {})} + + +class Follower: + """`--follow`: one connection at a time, opened again when it dies, and + every message printed once.""" + + def __init__(self, ctx: Context, account, group: dict, session, keys, check): + self.ctx, self.account, self.group = ctx, account, group + self.session, self.keys, self.check = session, keys, check + self.seen: set[str] = set() + self.last_ts = 0.0 + + def emit(self, m) -> None: + if m.key: + if m.key in self.seen: + return + self.seen.add(m.key) + self.last_ts = max(self.last_ts, m.timestamp) + if self.ctx.args.json: + print(json.dumps(as_json(m), ensure_ascii=False), flush=True) + else: + print_message(m, flush=True) + + async def run(self) -> None: + try: + while True: + await self._listen() + await self.session.close() + self.session = None + if not self.ctx.args.quiet: + err("meshbay-cli: the connection to the node was lost; reconnecting...") + await self._reopen() + finally: + if self.session is not None: + await self.session.close() + + async def _listen(self) -> None: + """Print what the node pushes until the connection dies.""" + s = self.session + queue: asyncio.Queue = asyncio.Queue() + dead = asyncio.Event() + + def on_message(msg): + queue.put_nowait(msg) + + def on_epoch(msg): + # Somebody was removed: the next message is sealed under a new key. + s.ack["chat_epoch"] = int(msg.get("epoch") or 0) + queue.put_nowait(None) + + def on_state(): + if s.pc is not None and s.pc.connectionState in ("failed", "closed"): + dead.set() + + s.ch.on_push("chat_msg", on_message) + s.ch.on_push("chat_epoch_ack", on_epoch) + if s.pc is not None: + s.pc.on("connectionstatechange", on_state) + pinger = asyncio.ensure_future(self._ping(dead)) + closed = asyncio.ensure_future(s.ch.closed.wait()) + gone = asyncio.ensure_future(dead.wait()) + try: + while True: + getter = asyncio.ensure_future(queue.get()) + done, _ = await asyncio.wait({getter, closed, gone}, + return_when=asyncio.FIRST_COMPLETED) + if getter not in done: + getter.cancel() + return + msg = getter.result() + if msg is None or int(msg.get("epoch") or 0) not in self.keys.by_epoch: + with contextlib.suppress(CliError): + self.keys = await chat_keys(s) + if msg is None: + continue + msg = {**msg, "timestamp": msg.get("timestamp") or 0} + self.emit(await open_row(msg, self.keys, self.group["id"], self.check)) + finally: + for task in (pinger, closed, gone): + task.cancel() + s.ch.off_push("chat_msg", on_message) + s.ch.off_push("chat_epoch_ack", on_epoch) + + async def _ping(self, dead: asyncio.Event) -> None: + """A connection that stopped answering looks open to both ends.""" + while True: + await asyncio.sleep(PING_EVERY) + try: + await self.session.ch.request({"type": "ping", "token": "follow"}, "pong", + timeout=PING_TIMEOUT) + except CliError: + dead.set() + return + + async def _reopen(self) -> None: + """A new connection, then what was said while there was none.""" + attempt = 0 + while True: + try: + self.session = await open_group(self.account, self.group) + self.keys = await chat_keys(self.session) + self.check = RosterCheck(self.account, self.session) + rows = await history(self.session, 200) + break + except CliError as e: + if self.session is not None: + await self.session.close() + self.session = None + if e.code != NODE: + raise + await asyncio.sleep(RECONNECT_WAITS[min(attempt, len(RECONNECT_WAITS) - 1)]) + attempt += 1 + for row in rows: + m = await open_row(row, self.keys, self.group["id"], self.check) + if m.timestamp >= self.last_ts: + self.emit(m) + + def safe_lines(text: str) -> list[str]: """A message's lines, each escaped: a line break stays a line break, every other control character is shown, never sent to the terminal.""" |