Source code for aiowsgi.thread

# -*- coding: utf-8 -*-
import time
import socket
import threading
from . import asyncio
from . import create_server
from six.moves import http_client


def get_free_port():
    s = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
    s.bind(('', 0))
    ip, port = s.getsockname()
    s.close()
    return ip, port


def check_server(host, port, path_info='/', timeout=3, retries=30):
    """Perform a request until the server reply"""
    if retries < 0:
        return 0
    time.sleep(.3)
    for i in range(retries):
        try:
            conn = http_client.HTTPConnection(host, int(port), timeout=timeout)
            conn.request('GET', path_info)
            res = conn.getresponse()
            return res.status
        except (socket.error, http_client.HTTPException):
            print('wait')
            time.sleep(.3)
    return 0


[docs]class WSGIServer(threading.Thread): """Stopable WSGI server running in a thread (not main thread). Usefull for functionnal testing. Usage: .. code-block:: >>> @asyncio.coroutine ... def application(environ, start_response): ... start_response('200 OK', []) ... return ['Hello world'] >>> loop = asyncio.get_event_loop() >>> server = WSGIServer(application) >>> server.start() >>> server.stop() ``server.url`` will contain the url to request """ def __init__(self, app, host='127.0.0.1', port=None): super(WSGIServer, self).__init__() self.server = None self.app = app _, self.port = port or get_free_port() self.host = host self.url = 'http://%s:%s' % (self.host, self.port) self.loop = asyncio.new_event_loop() def run(self): asyncio.set_event_loop(self.loop) self.server = create_server( self.app, loop=self.loop, host=self.host, port=self.port) self.server.run() def wait(self): info = (self.host, self.port) status = check_server(*info) if status not in (200, 399): self.loop.call_soon_threadsafe(self._stop) info += (status,) raise RuntimeError( 'Not able to connect to server at %s:%s (%s)' % info) def _stop(self): if self.server is not None: if getattr(self.server, 'aioserver', None) is not None: try: self.server.aioserver.close() except TypeError: pass self.server.close() self.loop.stop() def start(self): super(WSGIServer, self).start() self.wait() def stop(self): self.loop.call_soon_threadsafe(self._stop)