'Is using call_soon_threadsafe and GLib.idle_add together the correct way to safely call GTK code from asyncio, or is it overkill?
I have an application that runs in an asyncio event loop and I want to add a feature that interacts with my Gnome Shell, in particular sending a notification. Since GTK has its own (incompatible, I understand) event loop, I decided to throw the Gio.Application into a background thread. My question is about the appropriate way to run "GTK code" from my asyncio event loop, for example creating the notification and firing it from my Application.
Take the following code, I am scheduling the "add notification" method with the following line of code:
asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
app.add_notification, "You have a new email")
This "works", but I wonder if both call_soon_threadsafe and idle_add are redundant. It also works if I simply call
app.add_notification("You have a new email")
from my coroutine, however the PyGObject documentation suggests this is thread-unsafe. What's the recommended safe approach?
import asyncio
import threading
import gi
gi.require_version('Gio', '2.0')
from gi.repository import Gio, GLib
class ThreadsafeEvent(asyncio.Event):
def __init__(self, *args, **kwargs):
super().__init__(*args, **kwargs)
if self._loop is None:
self._loop = asyncio.get_event_loop()
def set(self):
self._loop.call_soon_threadsafe(super().set)
def clear(self):
self._loop.call_soon_threadsafe(super().clear)
class Application(Gio.Application):
APP_ID = "org.something"
def __init__(self, activated_event: threading.Event):
flags = Gio.ApplicationFlags.FLAGS_NONE
super().__init__(application_id=Application.APP_ID, flags=flags)
self.activated_event = activated_event
self.activated = False
def do_activate(self):
if not self.activated:
self.hold()
self.activated = True
self.activated_event.set()
def add_notification(self, message):
note = Gio.Notification.new(message)
self.send_notification(None, note)
async def main():
event = ThreadsafeEvent()
app = Application(event)
def start(app: Application):
app.register()
app.run()
thread = threading.Thread(target=start, args=(app,))
thread.setDaemon(True)
thread.start()
try:
# When the event is set, we know the Application
# is activated and safe to send notifications through
await asyncio.wait_for(event.wait(), 10)
except asyncio.TimeoutError:
print('Timed out before App received its activation.')
asyncio.get_event_loop().call_soon_threadsafe(GLib.idle_add,
app.add_notification, "You have a new email")
if __name__ == '__main__':
loop = asyncio.get_event_loop()
asyncio.ensure_future(main(), loop=loop)
loop.run_forever()
Sources
This article follows the attribution requirements of Stack Overflow and is licensed under CC BY-SA 3.0.
Source: Stack Overflow
| Solution | Source |
|---|
