tools/helpers/ipc.py

   1 # Copyright 2022 Alessandro Astone
   2 # SPDX-License-Identifier: GPL-3.0-or-later
   3
   4 # Currently implemented as FIFO
   5 import os
   6
   7 BASE_DIR = "/var/run/"
   8
   9 def pipe_for(channel):
  10     return BASE_DIR + "waydroid-" + channel
  11
  12 def read_one(channel):
  13     with open_channel(channel, "r", 1) as fifo:
  14         while True:
  15             data = fifo.read()
  16             if len(data) != 0:
  17                 return data
  18
  19 def create_channel(channel):
  20     pipe = pipe_for(channel)
  21     if not os.path.exists(pipe):
  22         os.mkfifo(pipe)
  23
  24 def open_channel(channel, mode, buffering=0):
  25     return open(pipe_for(channel), mode, buffering)
  26
  27 def notify(channel, msg):
  28     try:
  29         fd = os.open(pipe_for(channel), os.O_WRONLY | os.O_NONBLOCK)
  30         with os.fdopen(fd, "w") as fifo:
  31             fifo.write(msg)
  32     except Exception:
  33         pass
  34
  35 def notify_blocking(channel, msg):
  36     with open_channel(channel, "w", 1) as channel:
  37         channel.write(msg)