# Path: cc/reqs.py
# class TaskSendMessage(BaseMessage):
#     """Request to execute a task"""
#     req = Field(str, "task.send")
#     task_host = Field(str)
#     task_handler = Field(str)
#     task_id = Field(str)
# 
# Path: cc/stream.py
# class CCReqStream:
#     """Request-based API for CC socket.
# 
#     Add request-id into route, later map replies to original request
#     based on that.
#     """
# 
#     log = skytools.getLogger('CCReqStream')
# 
#     zmq_hwm = 100
#     zmq_linger = 500
# 
#     def __init__(self, cc_url, xtx, ioloop=None, zctx=None):
#         """Initialize stream."""
# 
#         zctx = zctx or zmq.Context.instance()
#         ioloop = ioloop or IOLoop.instance()
# 
#         s = zctx.socket (zmq.XREQ)
#         s.setsockopt (zmq.HWM, self.zmq_hwm)
#         s.setsockopt (zmq.LINGER, self.zmq_linger)
#         s.connect (cc_url)
# 
#         self.ccs = CCStream(s, ioloop, qmaxsize = self.zmq_hwm)
#         self.ioloop = ioloop
#         self.xtx = xtx
# 
#         self.query_id_seq = 1
#         self.query_cache = {}
# 
#         self.ccs.on_recv(self.handle_recv)
# 
#     def remove_query(self, qid):
#         """Drop query state.  Further replies are ignored."""
#         qi = self.query_cache.get(qid)
#         if qi:
#             del self.query_cache[qid]
#             qi.set_timeout(None)
# 
#     def ccquery_sync(self, msg, timeout=0):
#         """Synchronous query.
# 
#         Returns first reply.
#         """
#         res = [None]
#         def sync_cb(_rep):
#             res[0] = _rep
#             self.ioloop.stop()
#             return (False, 0)
#         self.ccquery_async(msg, sync_cb, timeout)
#         self.ioloop.start()
#         return res[0]
# 
#     def ccquery_async(self, msg, cbfunc, timeout=0):
#         """Asynchronous query.
# 
#         Maps replies to callback function based on request id.
#         """
#         # create query id prefix
#         qid = "Q%06d" % self.query_id_seq
#         self.query_id_seq += 1
# 
#         # create message, add query id
#         cmsg = self.xtx.create_cmsg(msg)
#         cmsg.set_route([qid])
# 
#         qi = QueryInfo(qid, cmsg, cbfunc, self)
#         self.query_cache[qid] = qi
# 
#         qi.set_timeout(timeout)
# 
#         qi.send_to(self.ccs)
# 
#         return qid
# 
#     def ccpublish(self, msg):
#         """Broadcast API."""
#         cmsg = self.xtx.create_cmsg(msg)
#         cmsg.send_to(self.ccs)
# 
#     def handle_recv(self, zmsg):
#         """Internal callback on ZMQStream.
# 
#         It must not throw exceptions.
#         """
#         try:
#             self.handle_recv_real(zmsg)
#         except Exception:
#             self.log.exception('handle_recv_real crashed, dropping msg: %r', zmsg)
# 
#     def handle_recv_real(self, zmsg):
#         """Actual callback that can throw exceptions."""
# 
#         cmsg = CCMessage(zmsg)
# 
#         route = cmsg.get_route()
#         if len(route) != 1:
#             self.log.error('Invalid reply route: %r', route)
#             return
# 
#         qid = route[0]
#         if qid not in self.query_cache:
#             self.log.error('reply for unknown query: %r', qid)
#             return
# 
#         msg = cmsg.get_payload(self.xtx)
# 
#         qi = self.query_cache[qid]
#         qi.launch_cb(msg)
# 
#     def resend(self, qid, timeout=0):
#         if qid in self.query_cache:
#             qi = self.query_cache[qid]
#             qi.send_to(self.ccs)
#             qi.set_timeout(timeout)
#         else:
#             pass # ?


import uuid
import skytools
from zmq.eventloop import IOLoop
from cc.reqs import TaskSendMessage
from cc.stream import CCReqStream
        elif tup[:2] == ['task', 'reply']:
            done = msg.status in ('finished', 'failed', 'stopped')
            self.log.debug('got result: %r', msg)
        else:
            done = False
            self.log.debug('got random: %r', msg)
        self.replies.append(msg)
        self.task_cbfunc(done, msg)
        self.log.debug('done=%r', done)
        if done:
            return (False, 0)
        return (True, 0)

class TaskManager:
    """Manages task on single CCReqStream connection."""

    log = skytools.getLogger('TaskManager')

    def __init__(self, ccrq):
        self.ccrq = ccrq
        self.ioloop = ccrq.ioloop

    def send_task_async(self, task, task_cbfunc):
        """Async task launch.

        Callback function will be called on replies.

        @param task: TaskSendMessage
        @param task_cbfunc: func with args (is_done, reply_msg)
        """

# Path: cc/reqs.py
# class TaskSendMessage(BaseMessage):
#     """Request to execute a task"""
#     req = Field(str, "task.send")
#     task_host = Field(str)
#     task_handler = Field(str)
#     task_id = Field(str)
# 
# Path: cc/stream.py
# class CCReqStream:
#     """Request-based API for CC socket.
# 
#     Add request-id into route, later map replies to original request
#     based on that.
#     """
# 
#     log = skytools.getLogger('CCReqStream')
# 
#     zmq_hwm = 100
#     zmq_linger = 500
# 
#     def __init__(self, cc_url, xtx, ioloop=None, zctx=None):
#         """Initialize stream."""
# 
#         zctx = zctx or zmq.Context.instance()
#         ioloop = ioloop or IOLoop.instance()
# 
#         s = zctx.socket (zmq.XREQ)
#         s.setsockopt (zmq.HWM, self.zmq_hwm)
#         s.setsockopt (zmq.LINGER, self.zmq_linger)
#         s.connect (cc_url)
# 
#         self.ccs = CCStream(s, ioloop, qmaxsize = self.zmq_hwm)
#         self.ioloop = ioloop
#         self.xtx = xtx
# 
#         self.query_id_seq = 1
#         self.query_cache = {}
# 
#         self.ccs.on_recv(self.handle_recv)
# 
#     def remove_query(self, qid):
#         """Drop query state.  Further replies are ignored."""
#         qi = self.query_cache.get(qid)
#         if qi:
#             del self.query_cache[qid]
#             qi.set_timeout(None)
# 
#     def ccquery_sync(self, msg, timeout=0):
#         """Synchronous query.
# 
#         Returns first reply.
#         """
#         res = [None]
#         def sync_cb(_rep):
#             res[0] = _rep
#             self.ioloop.stop()
#             return (False, 0)
#         self.ccquery_async(msg, sync_cb, timeout)
#         self.ioloop.start()
#         return res[0]
# 
#     def ccquery_async(self, msg, cbfunc, timeout=0):
#         """Asynchronous query.
# 
#         Maps replies to callback function based on request id.
#         """
#         # create query id prefix
#         qid = "Q%06d" % self.query_id_seq
#         self.query_id_seq += 1
# 
#         # create message, add query id
#         cmsg = self.xtx.create_cmsg(msg)
#         cmsg.set_route([qid])
# 
#         qi = QueryInfo(qid, cmsg, cbfunc, self)
#         self.query_cache[qid] = qi
# 
#         qi.set_timeout(timeout)
# 
#         qi.send_to(self.ccs)
# 
#         return qid
# 
#     def ccpublish(self, msg):
#         """Broadcast API."""
#         cmsg = self.xtx.create_cmsg(msg)
#         cmsg.send_to(self.ccs)
# 
#     def handle_recv(self, zmsg):
#         """Internal callback on ZMQStream.
# 
#         It must not throw exceptions.
#         """
#         try:
#             self.handle_recv_real(zmsg)
#         except Exception:
#             self.log.exception('handle_recv_real crashed, dropping msg: %r', zmsg)
# 
#     def handle_recv_real(self, zmsg):
#         """Actual callback that can throw exceptions."""
# 
#         cmsg = CCMessage(zmsg)
# 
#         route = cmsg.get_route()
#         if len(route) != 1:
#             self.log.error('Invalid reply route: %r', route)
#             return
# 
#         qid = route[0]
#         if qid not in self.query_cache:
#             self.log.error('reply for unknown query: %r', qid)
#             return
# 
#         msg = cmsg.get_payload(self.xtx)
# 
#         qi = self.query_cache[qid]
#         qi.launch_cb(msg)
# 
#     def resend(self, qid, timeout=0):
#         if qid in self.query_cache:
#             qi = self.query_cache[qid]
#             qi.send_to(self.ccs)
#             qi.set_timeout(timeout)
#         else:
#             pass # ?


# Path: cc/taskmgr.py
import uuid
import skytools
from zmq.eventloop import IOLoop
from cc.reqs import TaskSendMessage
from cc.stream import CCReqStream

        elif tup[:2] == ['task', 'reply']:
            done = msg.status in ('finished', 'failed', 'stopped')
            self.log.debug('got result: %r', msg)
        else:
            done = False
            self.log.debug('got random: %r', msg)
        self.replies.append(msg)
        self.task_cbfunc(done, msg)
        self.log.debug('done=%r', done)
        if done:
            return (False, 0)
        return (True, 0)

class TaskManager:
    """Manages task on single CCReqStream connection."""

    log = skytools.getLogger('TaskManager')

    def __init__(self, ccrq):
        self.ccrq = ccrq
        self.ioloop = ccrq.ioloop

    def send_task_async(self, task, task_cbfunc):
        """Async task launch.

        Callback function will be called on replies.

        @param task: TaskSendMessage
        @param task_cbfunc: func with args (is_done, reply_msg)
        """

        assert isinstance(task, TaskSendMessage)