# -*- coding: utf-8 -*-

import sys
from concurrent.futures import ThreadPoolExecutor
from tornado import gen, queues
from tornado.concurrent import Future, run_on_executor

class WorkItem(object):

    executor = ThreadPoolExecutor(20)

    def __init__(self, future, fn, args, kwargs):
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    def run(self):
        try:
            result = self.fn(*self.args, **self.kwargs)
        except Exception:
            self.future.set_exc_info(sys.exc_info())
        else:
            self.future.set_result(result)

class AsyncItem(object):

    executor = ThreadPoolExecutor(20)

    def __init__(self, future, fn, args, kwargs):
        self.future = future
        self.fn = fn
        self.args = args
        self.kwargs = kwargs

    @gen.coroutine
    def run(self):
        try:
            result = yield self.fn(*self.args, **self.kwargs)
        except Exception:
            self.future.set_exc_info(sys.exc_info())
        else:
            self.future.set_result(result)

class QueueExecutor(object):

    def __init__(self, pool_key, max_size, max_workers):
        self._pool_key = pool_key
        self._max_size = max_size
        self._max_workers = max_workers
        self._work_queue = queues.Queue(max_size)

    def submit(self, fn, *args, **kwargs):
        future = Future()
        item = AsyncItem(future, fn, args, kwargs)
        self._work_queue.put_nowait(item)
        return future

    @gen.coroutine
    def consume(self):
        while True:
            try:
                item = yield self._work_queue.get()
                yield item.run()
            finally:
                self._work_queue.task_done()

@@ -29,16 +29,17 @@
 ort sys%0A
+%0A
 from con
@@ -168,16 +168,78 @@
 ecutor%0A%0A
+from blackgate.http_client import fetch, handle_client_error%0A%0A
 class Wo
@@ -1219,21 +1219,8 @@
 size
-, max_workers
 ):%0A 
@@ -1290,48 +1290,8 @@
 ize%0A
-        self._max_workers = max_workers%0A
     
@@ -1707,16 +1707,16 @@
 inally:%0A
-
         
@@ -1748,12 +1748,1004 @@
 task_done()%0A
+%0A%0Aclass ExecutorPools(object):%0A%0A    def __init__(self):%0A        self.pools = %7B%7D%0A%0A    def register_pool(self, ioloop, group_key, max_size=10):%0A        executor = QueueExecutor(pool_key=group_key, max_size=max_size)%0A        ioloop.spawn_callback(executor.consume)%0A        self.pools%5Bgroup_key%5D = executor%0A        return executor%0A%0A    def get_executor(self, group_key):%0A        if group_key not in self.pools:%0A            raise Exception(%22Pool not registerd: %25s%22 %25 group_key)%0A        return self.pools%5Bgroup_key%5D%0A%0A    @gen.coroutine%0A    def execute(self, request, options=None):%0A        group_key = options.get('name')%0A        executor = self.get_executor(group_key)%0A        options = options or %7B%7D%0A        try:%0A            submit_data = dict(request=request, options=options)%0A            result = yield executor.submit(fetch, **submit_data)%0A            raise gen.Return(result)%0A        except Exception as exc:%0A            result = handle_client_error(exc)%0A            raise gen.Return(result)%0A

define executor pool in executor module.