def boot_frontend(config, debug=False):
    """
    Boot a Pyramid WSGI application as Twisted component
    """

    http_port = int(config.get('config-web', 'http_port'))
    websocket_uri = unicode(config.get('wamp', 'listen'))

    # https://stackoverflow.com/questions/13122519/serving-pyramid-application-using-twistd/13138610#13138610
    config = resource_filename('kotori.frontend', 'development.ini')
    application = get_app(config, 'main')

    # https://twistedmatrix.com/documents/13.1.0/web/howto/web-in-60/wsgi.html
    resource = WSGIResource(reactor, reactor.getThreadPool(), application)

    reactor.listenTCP(http_port, Site(resource)) 

def test_default(self):
        """
        When not otherwise initialized, the attribute evaluates to a
        ``_SyncToThreadedAsyncAPIAdapter`` using the global reactor, the global
        reactor's thread pool, and the value of ``block_device_api``.
        """
        threadpool = reactor.getThreadPool()

        api = UnusableAPI()
        deployer = BlockDeviceDeployer(
            hostname=u"192.0.2.1",
            node_uuid=uuid4(),
            block_device_api=api,
        )

        self.assertEqual(
            _SyncToThreadedAsyncAPIAdapter(
                _reactor=reactor, _threadpool=threadpool, _sync=api
            ),
            deployer.async_block_device_api,
        ) 

def setUp():
    for name in ('boto3', 'botocore'):
        logging.getLogger(name).setLevel(logging.CRITICAL)
    global ddb_process, boto_resource
    cmd = " ".join([
        "java", "-Djava.library.path=%s" % ddb_lib_dir,
        "-jar", ddb_jar, "-sharedDb", "-inMemory"
    ])
    ddb_process = subprocess.Popen(cmd, shell=True, env=os.environ)
    if os.getenv("AWS_LOCAL_DYNAMODB") is None:
        os.environ["AWS_LOCAL_DYNAMODB"] = "http://127.0.0.1:8000"
    boto_resource = DynamoDBResource()
    # Setup the necessary message tables
    message_table = os.environ.get("MESSAGE_TABLE", "message_int_test")
    create_rotating_message_table(prefix=message_table, delta=-1,
                                  boto_resource=boto_resource)
    create_rotating_message_table(prefix=message_table,
                                  boto_resource=boto_resource)
    pool = reactor.getThreadPool()
    pool.adjustPoolsize(minthreads=pool.max) 

def start(self, stop_after_crawl=True):
        """
        This method starts a Twisted `reactor`_, adjusts its pool size to
        :setting:`REACTOR_THREADPOOL_MAXSIZE`, and installs a DNS cache based
        on :setting:`DNSCACHE_ENABLED` and :setting:`DNSCACHE_SIZE`.

        If ``stop_after_crawl`` is True, the reactor will be stopped after all
        crawlers have finished, using :meth:`join`.

        :param boolean stop_after_crawl: stop or not the reactor when all
            crawlers have finished
        """
        if stop_after_crawl:
            d = self.join()
            # Don't start the reactor if the deferreds are already fired
            if d.called:
                return
            d.addBoth(self._stop_reactor)

        reactor.installResolver(self._get_dns_resolver())
        tp = reactor.getThreadPool()
        tp.adjustPoolsize(maxthreads=self.settings.getint('REACTOR_THREADPOOL_MAXSIZE'))
        reactor.addSystemEventTrigger('before', 'shutdown', self.stop)
        reactor.run(installSignalHandlers=False)  # blocking call 

def start(self, stop_after_crawl=True):
        """
        This method starts a Twisted `reactor`_, adjusts its pool size to
        :setting:`REACTOR_THREADPOOL_MAXSIZE`, and installs a DNS cache based
        on :setting:`DNSCACHE_ENABLED` and :setting:`DNSCACHE_SIZE`.

        If ``stop_after_crawl`` is True, the reactor will be stopped after all
        crawlers have finished, using :meth:`join`.

        :param boolean stop_after_crawl: stop or not the reactor when all
            crawlers have finished
        """
        if stop_after_crawl:
            d = self.join()
            # Don't start the reactor if the deferreds are already fired
            if d.called:
                return
            d.addBoth(self._stop_reactor)

        reactor.installResolver(self._get_dns_resolver())
        tp = reactor.getThreadPool()
        tp.adjustPoolsize(maxthreads=self.settings.getint('REACTOR_THREADPOOL_MAXSIZE'))
        reactor.addSystemEventTrigger('before', 'shutdown', self.stop)
        reactor.run(installSignalHandlers=False)  # blocking call 

def start_user_agent_in_single_user_mode(root_resource, services_factory, leap_home, leap_session):
    log.info('Bootstrap done, loading services for user %s' % leap_session.user_auth.username)

    _services = services.Services(leap_session)
    yield _services.setup()

    if leap_session.fresh_account:
        yield add_welcome_mail(leap_session.mail_store)

    services_factory.add_session(leap_session.user_auth.uuid, _services)

    authenticator = Authenticator(leap_session.provider)
    root_resource.initialize(provider=leap_session.provider, authenticator=authenticator)

    # soledad needs lots of threads
    reactor.getThreadPool().adjustPoolsize(5, 15)
    log.info('Done, the user agent is ready to be used') 

def test_configures_thread_pool(self):
        # Patch and restore where it's visible because patching a running
        # reactor is potentially fairly harmful.
        patcher = monkey.MonkeyPatcher()
        patcher.add_patch(reactor, "threadpool", None)
        patcher.add_patch(reactor, "threadpoolForDatabase", None)
        patcher.patch()
        try:
            service_maker = RegionMasterServiceMaker("Harry", "Hill")
            # Disable _ensureConnection() its not allowed in the reactor.
            self.patch_autospec(service_maker, "_ensureConnection")
            service_maker.makeService(Options())
            threadpool = reactor.getThreadPool()
            self.assertThat(threadpool, IsInstance(ThreadPool))
        finally:
            patcher.restore() 

def test_configures_thread_pool(self):
        # Patch and restore where it's visible because patching a running
        # reactor is potentially fairly harmful.
        patcher = monkey.MonkeyPatcher()
        patcher.add_patch(reactor, "threadpool", None)
        patcher.add_patch(reactor, "threadpoolForDatabase", None)
        patcher.patch()
        try:
            service_maker = RegionAllInOneServiceMaker("Harry", "Hill")
            # Disable _ensureConnection() its not allowed in the reactor.
            self.patch_autospec(service_maker, "_ensureConnection")
            service_maker.makeService(Options())
            threadpool = reactor.getThreadPool()
            self.assertThat(threadpool, IsInstance(ThreadPool))
        finally:
            patcher.restore() 

def deferToThread(f, *args, **kwargs):
    """
    Run a function in a thread and return the result as a Deferred.

    @param f: The function to call.
    @param *args: positional arguments to pass to f.
    @param **kwargs: keyword arguments to pass to f.

    @return: A Deferred which fires a callback with the result of f,
    or an errback with a L{twisted.python.failure.Failure} if f throws
    an exception.
    """
    from twisted.internet import reactor
    return deferToThreadPool(reactor, reactor.getThreadPool(),
                             f, *args, **kwargs) 

def from_api(cls, block_device_api, reactor=None):
        if reactor is None:
            from twisted.internet import reactor
        return cls(
            _sync=block_device_api,
            _reactor=reactor,
            _threadpool=reactor.getThreadPool(),
        ) 

def deferToThread(f, *args, **kwargs):
    """
    Run a function in a thread and return the result as a Deferred.

    @param f: The function to call.
    @param *args: positional arguments to pass to f.
    @param **kwargs: keyword arguments to pass to f.

    @return: A Deferred which fires a callback with the result of f,
    or an errback with a L{twisted.python.failure.Failure} if f throws
    an exception.
    """
    from twisted.internet import reactor
    return deferToThreadPool(reactor, reactor.getThreadPool(),
                             f, *args, **kwargs) 

def tearDown(self):
        # Unit tests that spawn a (blocking) client in a thread might still
        # have threads running at this point, if one is stuck waiting for a
        # message from a companion which has exited with an error. Our
        # relay's .stopService() drops all connections, which ought to
        # encourage those threads to terminate soon. If they don't, print a
        # warning to ease debugging.

        # XXX FIXME there's something in _noclobber test that's not
        # waiting for a close, I think -- was pretty relieably getting
        # unclean-reactor, but adding a slight pause here stops it...

        tp = reactor.getThreadPool()
        if not tp.working:
            yield self.sp.stopService()
            yield task.deferLater(reactor, 0.1, lambda: None)
            defer.returnValue(None)
        # disconnect all callers
        d = defer.maybeDeferred(self.sp.stopService)
        # wait a second, then check to see if it worked
        yield task.deferLater(reactor, 1.0, lambda: None)
        if len(tp.working):
            log.msg("wormhole.test.common.ServerBase.tearDown:"
                    " I was unable to convince all threads to exit.")
            tp.dumpStats()
            print("tearDown warning: threads are still active")
            print("This test will probably hang until one of the"
                  " clients gives up of their own accord.")
        else:
            log.msg("wormhole.test.common.ServerBase.tearDown:"
                    " I convinced all threads to exit.")
        yield d 

def deferToThread(f, *args, **kwargs):
    """
    Run a function in a thread and return the result as a Deferred.

    @param f: The function to call.
    @param *args: positional arguments to pass to f.
    @param **kwargs: keyword arguments to pass to f.

    @return: A Deferred which fires a callback with the result of f,
    or an errback with a L{twisted.python.failure.Failure} if f throws
    an exception.
    """
    from twisted.internet import reactor
    return deferToThreadPool(reactor, reactor.getThreadPool(),
                             f, *args, **kwargs) 

def _start_in_multi_user_mode(args, root_resource, services_factory):
    try:
        protected_resources = _setup_multi_user(args, root_resource, services_factory)
        start_site(args, protected_resources)
        reactor.getThreadPool().adjustPoolsize(5, 15)
        return defer.succeed(None)
    except Exception as e:
        return defer.fail(e) 

def test_configures_thread_pool(self):
        # Patch and restore where it's visible because patching a running
        # reactor is potentially fairly harmful.
        patcher = monkey.MonkeyPatcher()
        patcher.add_patch(reactor, "threadpool", None)
        patcher.add_patch(reactor, "threadpoolForDatabase", None)
        patcher.patch()
        try:
            service_maker = RegionWorkerServiceMaker("Harry", "Hill")
            service_maker.makeService(Options())
            threadpool = reactor.getThreadPool()
            self.assertThat(threadpool, IsInstance(ThreadPool))
        finally:
            patcher.restore() 

def start(self):
        """Start the Web Server """
        self.site = Site(WSGIResource(reactor, reactor.getThreadPool(), self.app))
        self.port = reactor.listenTCP(self.server.config.webport, self.site) 

def dump_stats():
    """
    Dump some basic stats about the reactor pool and threads at info level
    :return:
    """

    logger.info('Reactor queue stats: {}'.format(reactor.getThreadPool()._team.statistics().__dict__)) 

def make_icloudapi_tests(
        blockdevice_api_factory,
):
    """
    :param blockdevice_api_factory: A factory which will be called
        with the generated ``TestCase`` during the ``setUp`` for each
        test and which should return a provider of both ``IBlockDeviceAPI``
        and ``ICloudAPI`` to be tested.

    :returns: A ``TestCase`` with tests that will be performed on the
       supplied ``IBlockDeviceAPI``/``ICloudAPI`` provider.
    """
    class Tests(AsyncTestCase):
        def setUp(self):
            super(Tests, self).setUp()
            self.api = blockdevice_api_factory(test_case=self)
            self.this_node = self.api.compute_instance_id()
            self.async_cloud_api = _SyncToThreadedAsyncCloudAPIAdapter(
                _reactor=reactor, _sync=self.api,
                _threadpool=reactor.getThreadPool())

        def test_interface(self):
            """
            The result of the factory provides ``ICloudAPI``.
            """
            self.assertTrue(verifyObject(ICloudAPI, self.api))

        def test_current_machine_is_live(self):
            """
            The machine running the test is reported as alive.
            """
            d = self.async_cloud_api.list_live_nodes()
            d.addCallback(lambda live:
                          self.assertIn(self.api.compute_instance_id(), live))
            return d

        def test_list_live_nodes(self):
            """
            ``list_live_nodes`` returns an iterable of unicode values.
            """
            live_nodes = self.api.list_live_nodes()
            self.assertThat(live_nodes, AllMatch(IsInstance(unicode)))

    return Tests 

def prepare_twisted_service(handler, reactor_thread_size=100):
    """prepare twsited service
    """
    LOG.info('Prepare twisted services')

    LOG.info('Get peer configuration')
    for conf_key in CONF.bgp.running_config:
        LOG.info('---%s = %s', conf_key, CONF.bgp.running_config[conf_key])

    # init handler
    handler.init()

    LOG.info('Create BGPPeering twsited instance')
    afi_safi_list = [bgp_cons.AFI_SAFI_STR_DICT[afi_safi] for afi_safi in CONF.bgp.running_config['afi_safi']]
    CONF.bgp.running_config['afi_safi'] = afi_safi_list
    CONF.bgp.running_config['capability']['local']['afi_safi'] = afi_safi_list
    bgp_peering = BGPPeering(
        myasn=CONF.bgp.running_config['local_as'],
        myaddr=CONF.bgp.running_config['local_addr'],
        peerasn=CONF.bgp.running_config['remote_as'],
        peeraddr=CONF.bgp.running_config['remote_addr'],
        afisafi=CONF.bgp.running_config['afi_safi'],
        md5=CONF.bgp.running_config['md5'],
        handler=handler
    )
    CONF.bgp.running_config['factory'] = bgp_peering

    # Starting api server
    LOG.info("Prepare RESTAPI service")
    LOG.info("reactor_thread_size = %s", reactor_thread_size)
    reactor.suggestThreadPoolSize(reactor_thread_size)
    resource = WSGIResource(reactor, reactor.getThreadPool(), app)
    site = Site(resource)
    try:
        reactor.listenTCP(CONF.rest.bind_port, site, interface=CONF.rest.bind_host)
        LOG.info("serving RESTAPI on http://%s:%s", CONF.rest.bind_host, CONF.rest.bind_port)
    except Exception as e:
        LOG.error(e, exc_info=True)
        sys.exit()

    LOG.info('Starting BGPPeering twsited instance')
    bgp_peering.automatic_start()

    reactor.run()