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._joined_room_id = None
  12
  13     async def run(self, client):
  14         self._log(logging.INFO, "Presence running")
  15         self._stop = False
  16
  17         done, pending = await asyncio.wait(
  18             [asyncio.create_task( self._room_joining_loop(client) ),
  19              asyncio.create_task( self._message_loop(client) )],
  20             return_when = asyncio.FIRST_EXCEPTION
  21         )
  22
  23         for task in done:
  24             task.result()
  25
  26         self._log(logging.INFO, "Presence stopped")
  27
  28     def stop(self):
  29         asyncio.get_running_loop().call_soon_threadsafe(
  30             Presence._do_stop, self
  31         )
  32
  33     def _do_stop(self):
  34         self._stop = True
  35         self._tickle_fifo()
  36
  37     async def _room_joining_loop(self, client):
  38         while not self._stop:
  39             await client.sync()
  40
  41             prior_joined_room_id = self._joined_room_id
  42             self._joined_room_id = await self._get_joined_room_id(client)
  43
  44             if self._joined_room_id != None:
  45                 if self._joined_room_id != prior_joined_room_id:
  46                     self._log(logging.INFO, "Joined room")
  47             else:
  48                 if self._joined_room_id != prior_joined_room_id:
  49                     self._log(logging.INFO, "Got kicked out of room?")
  50                 else:
  51                     self._log(logging.INFO, "Not yet a room member")
  52                 was_invited = False
  53                 for room_id, room in client.invited_rooms.items():
  54                     if room.display_name == self._room_name:
  55                         was_invited = True
  56                         self._log(logging.INFO, "Joining room")
  57                         await client.join(room_id)
  58                         break
  59                 if not was_invited:
  60                     self._log(logging.INFO, "Got no room invite yet")
  61
  62             await asyncio.sleep(3.0)
  63
  64     async def _message_loop(self, client):
  65         if self._stop:
  66             return
  67         async for line in self._read_fifo(self._fifo_path):
  68             if self._stop:
  69                 break
  70             if self._joined_room_id != None:
  71                 await client.room_send(
  72                     room_id=self._joined_room_id,
  73                     message_type="m.room.message",
  74                     content={
  75                         "msgtype": "m.text",
  76                         "body": line.rstrip("\n")
  77                     }
  78                 )
  79             else:
  80                 self._log(logging.WARNING, "Dropping message cause no room joined")
  81
  82     def _tickle_fifo(self):
  83         fifo = -1
  84         try:
  85             fifo = posix.open(self._fifo_path, posix.O_WRONLY | posix.O_NONBLOCK)
  86             posix.write(fifo, "\n".encode())
  87         except OSError as e:
  88             if e.errno == errno.ENXIO:
  89                 pass
  90         finally:
  91             if fifo != -1:
  92                 posix.close(fifo)
  93
  94     async def _get_joined_room_id(self, client):
  95         joined_room_ids = ( await client.joined_rooms() ).rooms
  96         for room_id, room in client.rooms.items():
  97             if room.display_name == self._room_name:
  98                 if room_id in joined_room_ids:
  99                     return room_id
 100         return None
 101
 102     def _log(self, level, msg, *args, **kwargs):
 103         logging.log(level, "P{%s}: %s" % (self._room_name, msg), *args, **kwargs)
 104
 105     @staticmethod
 106     async def _read_fifo(file_path):
 107         while True:
 108             async with aiofiles.open(file_path) as fifo:
 109                 async for line in fifo:
 110                     yield(line)