How to use the txaio.use_asyncio function in txaio

#
# The above copyright notice and this permission notice shall be included in
# 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 txaio
txaio.use_asyncio()

import asyncio

import autobahn

from autobahn.asyncio.websocket import WebSocketServerProtocol, \
    WebSocketServerFactory

from autobahn.websocket.compress import PerMessageDeflateOffer, \
    PerMessageDeflateOfferAccept

# FIXME: streaming mode API is currently incompatible with permessage-deflate!
USE_STREAMING_TESTEE = False


class TesteeServerProtocol(WebSocketServerProtocol):

if True:
   import txaio

   def use_tx():
      "monkey-patched for doc-building"
      from txaio import tx
      txaio._use_framework(tx)

   def use_aio():
      "monkey-patched for doc-building"
      from txaio import aio
      txaio._use_framework(aio)

   txaio.use_twisted = use_tx
   txaio.use_asyncio = use_aio


extensions = [
   'sphinx.ext.autodoc',
   'sphinx.ext.doctest',
   'sphinx.ext.intersphinx',
   'sphinx.ext.viewcode',
   'sphinx.ext.ifconfig',
   'sphinx.ext.todo',
   'sphinxcontrib.spelling',
   'txsphinx' # only needed for Autobahn|Python
]

spelling_lang = 'en_US'
spelling_show_suggestions = False
spelling_word_list_filename = 'spelling_wordlist.txt'

import traceback

from asyncio import BaseProtocol

import txaio
import zmq
import zmq.asyncio

from eventify import Eventify
from eventify.drivers.base import BaseComponent
from eventify.persist import persist_event
from eventify.persist.constants import EVENT_DB_HOST, EVENT_DB_USER, EVENT_DB_PASS, \
    EVENT_DB_NAME


txaio.use_asyncio()
ctx = zmq.asyncio.Context.instance()


class Component(BaseComponent):
    """
    Handle subscribing to topics
    """
    log = logging.getLogger("eventify.drivers.zeromq")

    def __init__(self, config, handlers):
        self.config = config
        self.handlers = handlers

    def run(self):
        """
        start component

import math
import copy

from autobahn.util import public, _LazyHexFormatter
from autobahn.wamp.exception import ProtocolError, SerializationError, TransportLost
from autobahn.asyncio.util import peer2str, get_serializers
import txaio

__all__ = (
    'WampRawSocketServerProtocol',
    'WampRawSocketClientProtocol',
    'WampRawSocketServerFactory',
    'WampRawSocketClientFactory'
)

txaio.use_asyncio()

FRAME_TYPE_DATA = 0
FRAME_TYPE_PING = 1
FRAME_TYPE_PONG = 2

MAGIC_BYTE = 0x7F


class PrefixProtocol(asyncio.Protocol):

    prefix_format = '!L'
    prefix_length = struct.calcsize(prefix_format)
    max_length = 16 * 1024 * 1024
    max_length_send = max_length
    log = txaio.make_logger()  # @UndefinedVariable

# (usually, you can only import one or the other in a single Python
# interpreter)
import txaio

def use_tx():
  "monkey-patched for doc-building"
  from txaio import tx
  txaio._use_framework(tx)

def use_aio():
  "monkey-patched for doc-building"
  from txaio import aio
  txaio._use_framework(aio)

txaio.use_twisted = use_tx
txaio.use_asyncio = use_aio


# Monkey patch away WARNING like:
# "index.rst:4: WARNING: nonlocal image URI found: https://img.shields.io/pypi/v/txaio.svg"
# see: http://stackoverflow.com/a/28778969
import sphinx.environment
from docutils.utils import get_source_line

def _warn_node(self, msg, node, **kwargs):
    if not msg.startswith('nonlocal image URI found:'):
        self._warnfunc(msg, '%s:%s' % get_source_line(node))

sphinx.environment.BuildEnvironment.warn_node = _warn_node

try:
    from qualname import qualname

if parsed_url.scheme == 'tcp':
            is_unix = False
            if not parsed_url.hostname or not parsed_url.port:
                raise ValueError('Host and port is required in URL')
        elif parsed_url.scheme == 'unix' or parsed_url.scheme == '':
            is_unix = True
            if not parsed_url.path:
                raise ValueError('Path to unix socket must be in URL')

        transport_factory = WampRawSocketClientFactory(create, serializer=self.serializer)

        loop = asyncio.get_event_loop()
        if logging_level == 'debug':
            loop.set_debug(True)
        txaio.use_asyncio()
        txaio.config.loop = loop

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

        def handle_error(loop, context):
            self.log.error('Application Error: {err}', err=context)
            loop.stop()

        loop.set_exception_handler(handle_error)

        if is_unix:
            coro = loop.create_unix_connection(transport_factory, parsed_url.path)

import asyncio
import json
import logging
import os
import re
import sys
import traceback
import txaio

from aioimaplib import aioimaplib

from eventify import Eventify
from eventify.drivers.base import BaseComponent
from eventify.persist import persist_event

txaio.use_asyncio()

# You should probably use oauth but this driver
# is not meant for production use but only for
# a local tutorial
#
# If you wish to use this driver. Please note that
# you will need to enable "Less secure apps" on the
# google account you attempt to use.
GMAIL_USERNAME = os.getenv('GMAIL_USERNAME')
GMAIL_PASSWORD = os.getenv('GMAIL_PASSWORD')
IMAP_HOST = os.getenv('IMAP_HOST', 'imap.gmail.com')
SMTP_HOST = os.getenv('SMTP_HOST', 'smtp.gmail.com')
SMTP_PORT = os.getenv('SMTP_PORT', 465)
BODY_PATTERN = re.compile('Subject:[^{]*(.*)}')

async def start(self, ctx: Context):
        # Autobahn uses txaio to bridge the API gap between asyncio and Twisted so we need to set
        # it up for asyncio here
        txaio.use_asyncio()
        txaio.config.loop = ctx.loop

        ctx.add_resource(WAMPExtrasProvider(), 'wamp', types=[ExtrasProvider])

        for resource_name, context_attr, client in self.clients:
            await client.start(ctx)
            ctx.add_resource(client, resource_name, context_attr)
            logger.info('Configured WAMP client (%s / ctx.%s; host=%s; port=%d; realm=%s)',
                        resource_name, context_attr, client.host, client.port, client.realm)

        await yield_()

        for resource_name, context_attr, client in self.clients:
            await client.stop()
            logger.info('Shut down WAMP client (%s)', resource_name)

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

        # 2) create a WAMP-over-WebSocket transport client factory
        transport_factory = WampWebSocketClientFactory(create, url=self.url, serializers=self.serializers,
                                                       debug=self.debug, debug_wamp=self.debug_wamp)

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

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

        # 4) now enter the asyncio event loop
        try:
            loop.run_forever()
        except KeyboardInterrupt:
            # wait until we send Goodbye if user hit ctrl-c
            # (done outside this except so SIGTERM gets the same handling)

# 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

from collections import deque

import txaio
txaio.use_asyncio()

from autobahn.util import public
from autobahn.wamp import websocket
from autobahn.websocket import protocol

try:
    import asyncio
    from asyncio import iscoroutine
    from asyncio import Future
except ImportError:
    # Trollius >= 0.3 was renamed
    # noinspection PyUnresolvedReferences
    import trollius as asyncio
    from trollius import iscoroutine
    from trollius import Future