src/presence.py

   1 import aiofiles
   2 import asyncio
   3 import errno
   4 import logging
   5 import posix
   6
   7 class Presence:
   8     def __init__(self, room_name, fifo_path):
   9         self._room_name = room_name
  10         self._fifo_path = fifo_path
  11         self._is_member = False
  12
  13     async def run(self, client):
  14         self._log(logging.INFO, "Presence running")
  15         self._stop = False
  16         asyncio.create_task( self._do_watch(client) )
  17         while not self._stop:
  18             room_id = await self._enter_room(client)
  19             try:
  20                 await self._process_messages(client, room_id)
  21             except Exception as e:
  22                 self._log(logging.ERROR, "Error processing messages: %s", str(e))
  23         self._log(logging.INFO, "Presence stopped")
  24
  25     def stop(self):
  26         asyncio.get_running_loop().call_soon_threadsafe(
  27             Presence._do_stop, self
  28         )
  29
  30     def _do_stop(self):
  31         self._stop = True
  32         self._tickle_fifo()
  33
  34     async def _enter_room(self, client):
  35         while not self._stop:
  36             await client.sync()
  37
  38             room_id = await self._joined_room_id(client)
  39             if room_id != None:
  40                 self._log(logging.INFO, "A room member already")
  41                 self._is_member = True
  42                 return room_id
  43
  44             self._log(logging.INFO, "Not yet a room member")
  45
  46             was_invited = False
  47             for room_id, room in client.invited_rooms.items():
  48                 if room.display_name == self._room_name:
  49                     was_invited = True
  50                     self._log(logging.INFO, "Joining room")
  51                     await client.join(room_id)
  52                     break
  53
  54             if not was_invited:
  55                 self._log(logging.INFO, "Got no room invite yet")
  56
  57             await asyncio.sleep(3.0)
  58
  59     async def _do_watch(self, client):
  60         while not self._stop:
  61             self._log(logging.INFO, "WATCHING...")
  62
  63             await client.sync()
  64
  65             if await self._joined_room_id(client) == None:
  66                 self._is_member = False
  67                 self._tickle_fifo()
  68             else:
  69                 self._is_member = True
  70
  71             await asyncio.sleep(3.0)
  72
  73     def _tickle_fifo(self):
  74         fifo = -1
  75         try:
  76             fifo = posix.open(self._fifo_path, posix.O_WRONLY | posix.O_NONBLOCK)
  77             posix.write(fifo, "\n".encode())
  78         except OSError as e:
  79             if e.errno == errno.ENXIO:
  80                 pass
  81         finally:
  82             if fifo != -1:
  83                 posix.close(fifo)
  84
  85     async def _process_messages(self, client, room_id):
  86         if self._stop or not self._is_member:
  87             return
  88         async for line in self._read_fifo(self._fifo_path):
  89             if self._stop or not self._is_member:
  90                 break
  91             await client.room_send(
  92                 room_id=room_id,
  93                 message_type="m.room.message",
  94                 content={
  95                     "msgtype": "m.text",
  96                     "body": line.rstrip("\n")
  97                 }
  98             )
  99
 100     async def _joined_room_id(self, client):
 101         joined_room_ids = ( await client.joined_rooms() ).rooms
 102         for room_id, room in client.rooms.items():
 103             if room.display_name == self._room_name:
 104                 if room_id in joined_room_ids:
 105                     return room_id
 106         return None
 107
 108     def _log(self, level, msg, *args, **kwargs):
 109         logging.log(level, "P{%s}: %s" % (self._room_name, msg), *args, **kwargs)
 110
 111     @staticmethod
 112     async def _read_fifo(file_path):
 113         while True:
 114             async with aiofiles.open(file_path) as fifo:
 115                 async for line in fifo:
 116                     yield(line)