# -*- coding: utf-8 -*-
+import asyncio
import json
import logging
import re
from pika.adapters.asyncio_connection import AsyncioConnection
import slixmpp
+from slixmpp.exceptions import IqError, IqTimeout
from slixmpp.stanza import Message
from distbot.bot import action_worker
logging.getLogger('slixmpp').setLevel(logging.DEBUG)
logging.getLogger('slixmpp.xmlstream.xmlstream').setLevel(logging.DEBUG)
+# XEP-0410: how often to verify MUC membership, and the ping timeout (seconds).
+MUC_SELFPING_INTERVAL = 60
+MUC_SELFPING_TIMEOUT = 10
+# Error conditions from a MUC self-ping that mean "you are no longer joined".
+# Everything else (service-unavailable, feature-not-implemented,
+# remote-server-not-found/-timeout, an IqTimeout, ...) is inconclusive and must
+# NOT trigger a rejoin, so a flaky s2s link doesn't cause a rejoin storm.
+MUC_NOT_JOINED_CONDITIONS = frozenset({'not-acceptable', 'item-not-found'})
+
class Bot(slixmpp.ClientXMPP):
def __init__(self, jid, password, rooms, nick):
self._initialize_plugins()
self.muted = False
+ self._selfping_task = None
def _initialize_plugins(self):
self.register_plugin('xep_0045')
def disconnect(self, reconnect=False, wait=None, send_close=True):
logger.info("Stopping all workers...")
+ if self._selfping_task is not None:
+ self._selfping_task.cancel()
self.kill_workers()
logger.info("Stopping self...")
super(Bot, self).disconnect(reconnect)
ret = self.plugin['xep_0045'].join_muc(room, self.nick)
logger.info('%s: joined with code %s' % (room, ret))
+ # Start the XEP-0410 self-ping loop once. It survives across c2s
+ # reconnects, and — unlike session_start — also recovers from a silent
+ # MUC eviction (e.g. an s2s flap) that never touches the c2s stream.
+ if self._selfping_task is None or self._selfping_task.done():
+ self._selfping_task = asyncio.ensure_future(self._muc_selfping_loop())
+
+ async def _muc_selfping_loop(self):
+ """XEP-0410: periodically confirm we are still joined, rejoin if not."""
+ while True:
+ try:
+ await asyncio.sleep(MUC_SELFPING_INTERVAL)
+ for room in self.rooms:
+ await self._muc_selfping(room)
+ except asyncio.CancelledError:
+ raise
+ except Exception as e: # never let the loop die
+ logger.exception("MUC self-ping loop error: %s", e)
+
+ async def _muc_selfping(self, room):
+ """Ping our own occupant JID; rejoin if the MUC says we're not in it."""
+ occupant = '%s/%s' % (room, self.nick)
+ try:
+ await self.plugin['xep_0199'].ping(jid=occupant, timeout=MUC_SELFPING_TIMEOUT)
+ # An IQ result means the MUC routed the ping to us: still joined.
+ except IqError as e:
+ if e.condition in MUC_NOT_JOINED_CONDITIONS:
+ logger.warning("%s: self-ping says not joined (%s); rejoining",
+ room, e.condition)
+ self.plugin['xep_0045'].join_muc(room, self.nick)
+ else:
+ logger.debug("%s: self-ping inconclusive (%s); not rejoining",
+ room, e.condition)
+ except IqTimeout:
+ logger.debug("%s: self-ping timed out; assuming still joined", room)
+
def muc_message(self, msg: Message):
if msg['mucnick'] == self.nick or 'groupchat' != msg['type']:
return False