How to use the txaio.use_twisted function in txaio

def test_use_twisted(framework_tx):
    pytest.importorskip('twisted')

    import txaio
    txaio.use_twisted()
    assert txaio.using_twisted
    assert not txaio.using_asyncio

#
#  This program is free software: you can redistribute it and/or modify it under the
#  terms of the GNU Affero General Public License, version 3, as published by the
#  Free Software Foundation. This program is distributed in the hope that it will be
#  useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
#  See the GNU Affero General Public License Version 3 for more details.
#
#  You should have received a copy of the GNU Affero General Public license along
#  with this program. If not, see .
#
#####################################################################################

import txaio
txaio.use_twisted()

from txaio import make_logger

from pytrie import StringTrie

from zope.interface import implementer

from collections import OrderedDict

from twisted.internet.interfaces import IHandshakeListener, ISSLTransport
from twisted.internet.protocol import Protocol, Factory
from twisted.internet.defer import inlineCallbacks, Deferred, returnValue, succeed

from crossbar.bridge.mqtt.tx import MQTTServerTwistedProtocol
from crossbar.router.session import RouterSession

# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
###############################################################################

from __future__ import absolute_import

import txaio
txaio.use_twisted()

from twisted.python import usage
from twisted.internet.defer import inlineCallbacks
from twisted.internet.protocol import Factory, Protocol
from twisted.internet.endpoints import clientFromString, serverFromString
from twisted.application import service


class DestEndpointForwardingProtocol(Protocol):

    log = txaio.make_logger()

    def connectionMade(self):
        self.log.debug("DestEndpointForwardingProtocol.connectionMade")
        pass

# all copies or substantial portions of the Software.
#
# THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
# IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
# FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
###############################################################################

import sys
import txaio

txaio.use_twisted()

from twisted import logger
from twisted.logger import ILogObserver
from zope.interface import provider

# some library you use is using txaio logging stuff
class Library(object):
    log = txaio.make_logger()

    def something(self):
        self.log.info("info log from library foo={foo}", foo='bar')
        self.log.debug("debug information")

# lets say you start your own observer
@provider(ILogObserver)
class Observer(object):

from __future__ import print_function

import json
import binascii

import txaio
txaio.use_twisted()

from twisted.internet.defer import Deferred, succeed, inlineCallbacks, returnValue, CancelledError
from twisted.internet import protocol
from twisted.web.client import Agent, HTTPConnectionPool
from twisted.web.iweb import IBodyProducer, UNKNOWN_LENGTH
from twisted.web.http_headers import Headers

import six

import treq


__all__ = (
    'Client',
    'Value',
    'Header',

stops. If there are any problems starting the reactor or
           connect()-ing, we stop the reactor and raise the exception
           back to the caller.

        :returns: None is returned, unless you specify
            ``start_reactor=False`` in which case the Deferred that
            connect() returns is returned; this will callback() with
            an IProtocol instance, which will actually be an instance
            of :class:`WampWebSocketClientProtocol`
        """
        if start_reactor:
            # only select framework, set loop and start logging when we are asked
            # start the reactor - otherwise we are running in a program that likely
            # already tool care of all this.
            from twisted.internet import reactor
            txaio.use_twisted()
            txaio.config.loop = reactor
            txaio.start_logging(level=log_level)

        if callable(make):
            # factory for use ApplicationSession
            def create():
                cfg = ComponentConfig(self.realm, self.extra)
                try:
                    session = make(cfg)
                except Exception:
                    self.log.failure('ApplicationSession could not be instantiated: {log_failure.value}')
                    if start_reactor and reactor.running:
                        reactor.stop()
                    raise
                else:
                    return session

# AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
# LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
# OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
# THE SOFTWARE.
#
###############################################################################

from __future__ import absolute_import

from base64 import b64encode, b64decode
import six

from zope.interface import implementer

import txaio
txaio.use_twisted()

import twisted.internet.protocol
from twisted.internet import endpoints
from twisted.internet.interfaces import ITransport, ISSLTransport

from twisted.internet.error import ConnectionDone, ConnectionAborted, \
    ConnectionLost
from twisted.internet.defer import Deferred

from autobahn.util import public
from autobahn.util import _is_tls_error, _maybe_tls_reason
from autobahn.wamp import websocket
from autobahn.websocket.types import ConnectionRequest, ConnectionResponse, ConnectionDeny, \
    TransportDetails
from autobahn.websocket import protocol
from autobahn.websocket.interfaces import IWebSocketClientAgent

#  Free Software Foundation. This program is distributed in the hope that it will be
#  useful, but WITHOUT ANY WARRANTY; without even the implied warranty of
#  MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE.
#
#  See the GNU Affero General Public License Version 3 for more details.
#
#  You should have received a copy of the GNU Affero General Public license along
#  with this program. If not, see .
#
#####################################################################################

import time
from collections.abc import Mapping

import txaio
txaio.use_twisted()

import crossbar
from crossbar.common import checkconfig
from crossbar.node import node
from crossbar.node.worker import RouterWorkerProcess, ContainerWorkerProcess, WebSocketTesteeWorkerProcess
from crossbar.worker.router import RouterController
from crossbar.worker import transport
from crossbar.worker.container import ContainerController
from crossbar.worker.testee import WebSocketTesteeController
from crossbar.worker.proxy import ProxyController, ProxyWorkerProcess
from crossbar.webservice import base
from crossbar.webservice import wsgi, rest, longpoll, websocket, misc, static, archive, wap
from crossbar.router.realmstore import MemoryRealmStore


def do_nothing(*args, **kw):

import sys
import logging
import signal
import json
import base64
from datetime import timedelta

from twisted.internet import reactor
from twisted.web.server import Site
from twisted.web.resource import Resource
from autobahn.twisted.websocket import WebSocketServerFactory, \
    WebSocketServerProtocol
from autobahn.twisted.resource import WebSocketResource

import txaio
txaio.use_twisted()

DELTA = timedelta(weeks=1)
SAFE_WAIT = 3
START_COUNT = 10
VALID_AUTH_TOKENS = ["asdf", "asdf1", "asdf2", "asdf3"]


def websocket_func(logger, host, port):
    cons = list()

    # noinspection PyUnusedLocal
    def sigint_handler(signum, frame):
        reactor.stop()

    signal.signal(signal.SIGINT, sigint_handler)