ydb/contrib/python/Twisted/py3/twisted/internet/gireactor.py

# Copyright (c) Twisted Matrix Laboratories.
# See LICENSE for details.

"""
This module provides support for Twisted to interact with the glib
mainloop via GObject Introspection.

In order to use this support, simply do the following::

    from twisted.internet import gireactor
    gireactor.install()

If you wish to use a GApplication, register it with the reactor::

    from twisted.internet import reactor
    reactor.registerGApplication(app)

Then use twisted.internet APIs as usual.

On Python 3, pygobject v3.4 or later is required.
"""


from typing import Union

from gi.repository import GLib

from twisted.internet import _glibbase
from twisted.internet.error import ReactorAlreadyRunning
from twisted.python import runtime

if getattr(GLib, "threads_init", None) is not None:
    GLib.threads_init()


class GIReactor(_glibbase.GlibReactorBase):
    """
    GObject-introspection event loop reactor.

    @ivar _gapplication: A C{Gio.Application} instance that was registered
        with C{registerGApplication}.
    """

    # By default no Application is registered:
    _gapplication = None

    def __init__(self, useGtk=False):
        _glibbase.GlibReactorBase.__init__(self, GLib, None)

    def registerGApplication(self, app):
        """
        Register a C{Gio.Application} or C{Gtk.Application}, whose main loop
        will be used instead of the default one.

        We will C{hold} the application so it doesn't exit on its own. In
        versions of C{python-gi} 3.2 and later, we exit the event loop using
        the C{app.quit} method which overrides any holds. Older versions are
        not supported.
        """
        if self._gapplication is not None:
            raise RuntimeError("Can't register more than one application instance.")
        if self._started:
            raise ReactorAlreadyRunning(
                "Can't register application after reactor was started."
            )
        if not hasattr(app, "quit"):
            raise RuntimeError(
                "Application registration is not supported in"
                " versions of PyGObject prior to 3.2."
            )
        self._gapplication = app

        def run():
            app.hold()
            app.run(None)

        self._run = run

        self._crash = app.quit


class PortableGIReactor(_glibbase.GlibReactorBase):
    """
    Portable GObject Introspection event loop reactor.
    """

    def __init__(self, useGtk=False):
        super().__init__(GLib, None, useGtk=useGtk)

    def registerGApplication(self, app):
        """
        Register a C{Gio.Application} or C{Gtk.Application}, whose main loop
        will be used instead of the default one.
        """
        raise NotImplementedError("GApplication is not currently supported on Windows.")

    def simulate(self) -> None:
        """
        For compatibility only. Do nothing.
        """


def install(useGtk: bool = False) -> Union[GIReactor, PortableGIReactor]:
    """
    Configure the twisted mainloop to be run inside the glib mainloop.

    @param useGtk: A hint that the Gtk GUI will or will not be used.  Currently
        does not modify any behavior.
    """
    reactor: Union[GIReactor, PortableGIReactor]
    if runtime.platform.getType() == "posix":
        reactor = GIReactor(useGtk=useGtk)
    else:
        reactor = PortableGIReactor(useGtk=useGtk)

    from twisted.internet.main import installReactor

    installReactor(reactor)
    return reactor


__all__ = ["install"]