src/presence.py

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