<gh_stars>0
# -*- coding: utf-8 -*-

"""Main module."""

import eventlet
from concurrent.futures import Executor, Future, ThreadPoolExecutor
from functools import partial
from typing import List, Union
from eventlet.semaphore import BoundedSemaphore

_executor: Union[Executor, None] = None


def set_default_executor(executor: Executor):
    global _executor

    if executor is None:
        return

    if _executor is not None:
        del _executor

    _executor = executor


def run(func, *args, executor: Union[Executor, None] = None):
    """Arrange for a func to be called in the specified executor.

    The executor argument should be an Executor instance. The default executor
    is used if executor is None.

    Use functools.partial to pass keywords to the *func*.

    """
    global _executor

    if executor is None:
        if _executor is None:
            executor = ThreadPoolExecutor()
            _executor = executor
        else:
            executor = _executor

    return executor.submit(func, *args)


def wait(future: Future):
    """Wait future object finished.

    :param future: Future object generated by run_in_executor*()
    """
    sem = BoundedSemaphore()
    sem.acquire()
    future.add_done_callback(
        partial(lambda future, sem: sem.release(), sem=sem)
    )
    with sem:
        return


def wait_all(futures: List[Future]):
    """Wait all futures done

    :param futures: Futures list
    """
    sem = BoundedSemaphore(len(futures))
    for future in futures:
        sem.acquire()
        future.add_done_callback(
            partial(lambda future, sem: sem.release(), sem=sem)
        )

    while True:
        try:
            sem.release(blocking=False)
            sem.acquire()
            eventlet.sleep(0.05)
        except ValueError:
            break


def run_wait(func, *args, executor: Union[Executor, None] = None):
    future = run(func, *args, executor=executor)
    wait(future)
    return future
