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
  12     async def run(self, client):
  13         self._log(logging.INFO, "Presence running")
  14         self._stop = False
  15         while not self._stop:
  16             room_id = await self._enter_room(client)
  17             try:
  18                 await self._process_messages(client, room_id)
  19             except Exception as e:
  20                 self._log(logging.ERROR, "Error processing messages: %s", str(e))
  21         self._log(logging.INFO, "Presence stopped")
  22
  23     def stop(self):
  24         self._stop = True
  25         fifo = -1
  26         try:
  27             fifo = posix.open(self._fifo_path, posix.O_WRONLY | posix.O_NONBLOCK)
  28             posix.write(fifo, "\n".encode())
  29         except OSError as e:
  30             if e.errno == errno.ENXIO:
  31                 pass
  32         finally:
  33             if fifo != -1:
  34                 posix.close(fifo)
  35
  36     async def _enter_room(self, client):
  37         while not self._stop:
  38             await client.sync()
  39
  40             for room_id, room in client.rooms.items():
  41                 if room.display_name == self._room_name:
  42                     self._log(logging.INFO, "A room member already")
  43                     return room_id
  44
  45             self._log(logging.INFO, "Not yet a room member")
  46
  47             was_invited = False
  48             for room_id, room in client.invited_rooms.items():
  49                 if room.display_name == self._room_name:
  50                     was_invited = True
  51                     self._log(logging.INFO, "Joining room")
  52                     await client.join(room_id)
  53                     break
  54
  55             if not was_invited:
  56                 self._log(logging.INFO, "Got no room invite yet")
  57
  58             await asyncio.sleep(3.0)
  59
  60     async def _process_messages(self, client, room_id):
  61         if self._stop:
  62             return
  63         async for line in self._read_fifo(self._fifo_path):
  64             if self._stop:
  65                 break
  66             await client.room_send(
  67                 room_id=room_id,
  68                 message_type="m.room.message",
  69                 content={
  70                     "msgtype": "m.text",
  71                     "body": line.rstrip("\n")
  72                 }
  73             )
  74
  75     def _log(self, level, msg, *args, **kwargs):
  76         logging.log(level, "P{%s}: %s" % (self._room_name, msg), *args, **kwargs)
  77
  78     @staticmethod
  79     async def _read_fifo(file_path):
  80         while True:
  81             async with aiofiles.open(file_path) as fifo:
  82                 async for line in fifo:
  83                     yield(line)