How to use the txaio.config.loop function in txaio

from twisted.internet import task
        with replace_loop(task.Clock()) as fake_reactor:
            f = txaio.call_later(5, foo)
            fake_reactor.advance(10)
            # ...etc
    """

    # setup
    orig = txaio.config.loop
    txaio.config.loop = new_loop

    yield new_loop

    # cleanup
    txaio.config.loop = orig

from twisted.internet import task
        with replace_loop(task.Clock()) as fake_reactor:
            f = txaio.call_later(5, foo)
            fake_reactor.advance(10)
            # ...etc
    """

    # setup
    orig = txaio.config.loop
    txaio.config.loop = new_loop

    yield new_loop

    # cleanup
    txaio.config.loop = orig

event_loop or a reactor instance depending upon asyncio/Twisted.

    Use like so:

    .. sourcecode:: python

        from twisted.internet import task
        with replace_loop(task.Clock()) as fake_reactor:
            f = txaio.call_later(5, foo)
            fake_reactor.advance(10)
            # ...etc
    """

    # setup
    orig = txaio.config.loop
    txaio.config.loop = new_loop

    yield new_loop

    # cleanup
    txaio.config.loop = orig

pending Futures. This is a no-op for Twisted, because you don't
    need to use the event-loop to get callbacks to happen in Twisted.
    '''

    import txaio
    if txaio.using_twisted:
        return

    try:
        import asyncio
        if sys.version_info >= (3, 7):
            # https://github.com/crossbario/txaio/issues/139
            from _asyncio_test_utils import run_once as _run_once
        else:
            from asyncio.test_utils import run_once as _run_once
        return _run_once(txaio.config.loop or asyncio.get_event_loop())

    except ImportError:
        import trollius as asyncio
        # let any trollius import error out; if we're not using
        # twisted, and have no asyncio *and* no trollius, that's a
        # problem.

        # copied from asyncio.testutils because trollius has no
        # testutils"

        # just like modern asyncio.testutils.run_once does it...
        loop = asyncio.get_event_loop()
        loop.stop()
        loop.run_forever()
        asyncio.gather(*asyncio.Task.all_tasks())

from pprint import pformat
import txaio
from autobahn.asyncio.wamp import ApplicationSession

from .common import *
from ..environment import Environment
from ..exceptions import NoDriverFoundError, NoResourceFoundError, InvalidConfigError
from ..resource.remote import RemotePlaceManager, RemotePlace
from ..util.dict import diff_dict, flat_dict, filter_dict
from ..util.yaml import dump
from .. import Target, target_factory
from ..util.proxy import proxymanager
from ..util.helper import processwrapper

txaio.use_asyncio()
txaio.config.loop = asyncio.get_event_loop()


class Error(Exception):
    pass


class UserError(Error):
    pass


class ServerError(Error):
    pass


class ClientSession(ApplicationSession):
    """The ClientSession encapsulates all the actions a Client can Invoke on

def stop_server():
    """Shuts down all threads and exits cleanly."""
    global __server
    __server.shutdown()
    event_loop = txaio.config.loop
    event_loop.stop()
    # We've told the event loop to stop, but it won't shut down until we poke
    # it with a simple scheduled task.
    event_loop.call_soon_threadsafe(lambda: None)

    # If we are in spyder, undo our import. This gets done in the websocket
    # server onClose above if the browser tab is closed but is not done
    # if the user stops the kernel instead.
    if _in_spyder:
        _undo_vpython_import_in_spyder()

    # We don't want Ctrl-C to try to sys.exit inside spyder, i.e.
    # in an ipython console with a separate python kernel running.
    if _in_spyder:
        raise KeyboardInterrupt

perMessageCompressionAccept=accept)
        # SSL context for client connection
        if self.ssl is None:
            ssl = isSecure
        else:
            if self.ssl and not isSecure:
                raise RuntimeError(
                    'ssl argument value passed to %s conflicts with the "ws:" '
                    'prefix of the url argument. Did you mean to use "wss:"?' %
                    self.__class__.__name__)
            ssl = self.ssl

        # start the client connection
        loop = asyncio.get_event_loop()
        txaio.use_asyncio()
        txaio.config.loop = loop
        coro = loop.create_connection(transport_factory, host, port, ssl=ssl)

        # start a asyncio loop
        if not start_loop:
            return coro
        else:
            (transport, protocol) = loop.run_until_complete(coro)

            # start logging
            txaio.start_logging(level=log_level)

            try:
                loop.add_signal_handler(signal.SIGTERM, loop.stop)
            except NotImplementedError:
                # signals are not available on Windows
                pass