def test_get_callback_consumers(self):
        ps = RedisPubSub(self.kvdb, self.key_prefix)

        msg_value = '"msg_value"'

        topic = Topic('/test/delete')
        ps.add_topic(topic)

        producer = Client('Producer', 'producer')
        ps.add_producer(producer, topic)

        id1 = 'Consumer CB1'
        name1 = 'consumer-cb1'
        sub_key1 = new_cid()
        callback_id1 = rand_int()

        id2 = 'Consumer CB2'
        name2 = 'consumer-cb2'
        sub_key2 = new_cid()
        callback_id2 = rand_int()

        consumer_cb1 = Consumer(id1, name1, sub_key=sub_key1, delivery_mode=PUB_SUB.DELIVERY_MODE.CALLBACK_URL.id,
            callback_id=callback_id1)

        consumer_cb2 = Consumer(id2, name2, sub_key=sub_key2, delivery_mode=PUB_SUB.DELIVERY_MODE.CALLBACK_URL.id,
            callback_id=callback_id2)

        consumer_pull = Consumer('Consumer pull', 'consumer-pull', sub_key=new_cid(), delivery_mode=PUB_SUB.DELIVERY_MODE.PULL.id)
        consumer_inactive = Consumer(
            'Consumer pull', 'consumer-pull', is_active=False, sub_key=new_cid(), delivery_mode=PUB_SUB.DELIVERY_MODE.PULL.id)

        ps.add_consumer(consumer_cb1, topic)
        ps.add_consumer(consumer_cb2, topic)
        ps.add_consumer(consumer_pull, topic)     # This one should not be returned because it's a pull one
        ps.add_consumer(consumer_inactive, topic) # This one should not be returned because it's inactive

        consumers = list(ps.get_callback_consumers())

        # Only 2 are returned, the rest won't make it
        eq_(len(consumers), 2)

        # Sort by each consumer's ID, i.e. in lexicographical order
        consumers.sort(key=attrgetter('id'))

        consumer1 = consumers[0]
        eq_(consumer1.id, id1)
        eq_(consumer1.name, name1)
        eq_(consumer1.is_active, True)
        eq_(consumer1.sub_key, sub_key1)
        eq_(consumer1.callback_id, callback_id1)

        consumer2 = consumers[1]
        eq_(consumer2.id, id2)
        eq_(consumer2.name, name2)
        eq_(consumer2.is_active, True)
        eq_(consumer2.sub_key, sub_key2)
        eq_(consumer2.callback_id, callback_id2)

 def __init__(self, kvdb, client_type, topic_callbacks):
     Thread.__init__(self)
     self.kvdb = kvdb
     self.decrypt_func = kvdb.decrypt_func
     self.name = '{}-{}'.format(client_type, new_cid())
     self.topic_callbacks = topic_callbacks
     self._to_parallel_any_topic = TOPICS[MESSAGE_TYPE.TO_PARALLEL_ANY]

    def test_invoke_retry_exception_has_async(self):

        target = 'target_{}'.format(rand_string())
        callback = 'callback_{}'.format(rand_string())
        callback_impl_name = 'callback_impl_name_{}'.format(rand_string())
        cid = new_cid()
        expected_result = rand_string()

        invoking_service = DummyInvokingService(callback, callback_impl_name, cid, expected_result, raise_on_invoke=True)
        ir = InvokeRetry(invoking_service)

        kwargs = {
            'async_fallback': True,
            'callback': callback,
            'context': {rand_string():rand_string()},
            'repeats': rand_int(1, 10),
            'seconds': 0.01,
            'minutes': 0,
        }

        kwargs_copy = deepcopy(kwargs)

        try:
            ir.invoke_retry(target, 1, 2, 3, **kwargs)
        except NeedsRetry, e:
            self.assertEquals(e.cid, cid)
            self.assertEquals(e.cid, e.inner_exc.message)

    def get_data(self, data_format, transport, add_string=NON_ASCII_STRING, needs_payload=True,
            payload='', service_class=DummyAdminService):
        handler = channel.RequestHandler(get_dummy_server())

        expected = {
            'key': 'a' + uuid4().hex + add_string,
            'value': uuid4().hex + NON_ASCII_STRING,
            'result': uuid4().hex,
            'details': uuid4().hex,
            'cid': new_cid(),
            'zato':zato_namespace
        }

        if needs_payload:
            if not payload:
                if data_format == SIMPLE_IO.FORMAT.JSON:
                    payload_value = {expected['key']: expected['value']}
                else:
                    # NOTE: str.format can't handle Unicode arguments http://bugs.python.org/issue7300
                    payload_value = """<%(key)s xmlns="%(zato)s">%(value)s<zato_env>
                          <cid>%(cid)s</cid>
                          <result>%(result)s</result>
                          <details>%(details)s</details>
                        </zato_env>
                      </%(key)s>""" % (expected)
                payload = DummyPayload(payload_value)
        else:
            payload = None

        response = DummyResponse(payload, expected['result'], expected['details'])
        service = service_class(response, expected['cid'])

        handler.set_payload(response, data_format, transport, service)

        return expected, service

    def __init__(
        self,
        payload="",
        topic=None,
        mime_type=PUB_SUB.DEFAULT_MIME_TYPE,
        priority=PUB_SUB.DEFAULT_PRIORITY,
        expiration=PUB_SUB.DEFAULT_EXPIRATION,
        msg_id=None,
        producer=None,
        creation_time_utc=None,
        expire_at_utc=None,
    ):
        self.payload = payload
        self.topic = topic
        self.mime_type = mime_type
        self.priority = priority  # In 1-9 range where 9 is top priority
        self.msg_id = msg_id or new_cid()
        self.producer = producer
        self.expiration = expiration
        self.creation_time_utc = creation_time_utc or datetime.utcnow()
        self.expire_at_utc = expire_at_utc or (self.creation_time_utc + timedelta(seconds=self.expiration))

        # These two, in local timezone, are used by web-admin.
        self.creation_time = None
        self.expire_at = None

        self.id = None  # Used by frontend only
        self.payload_html = None  # Used by frontend only

    def test_client_ok(self):

        cid = new_cid()
        headers = {'x-zato-cid':cid}
        ok = True
        _rand = rand_string()
        soap_action = rand_string()
        
        text = """
            <soapenv:Envelope xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">
             <soapenv:Body>
              <abc>{}</abc>
             </soapenv:Body>
            </soapenv:Envelope>""".format(_rand).strip()
        status_code = rand_int()
        
        client = self.get_client(FakeInnerResponse(headers, ok, text, status_code))
        response = client.invoke(soap_action)

        expected_response_data = """
            <abc xmlns:soapenv="http://schemas.xmlsoap.org/soap/envelope/">{}</abc>
            """.format(_rand).strip()

        eq_(response.details, None)
        eq_(response.ok, ok)
        eq_(response.inner.text, text)
        eq_(etree.tostring(response.data), expected_response_data)
        eq_(response.has_data, True)
        eq_(response.cid, cid)

    def handle(self):
        input = self.request.input
        self._validate_input(input)

        with closing(self.odb.session()) as session:
            try:
                # Find a topic by its name so it can be paired with client_id later on
                topic = session.query(PubSubTopic).\
                    filter(PubSubTopic.cluster_id==input.cluster_id).\
                    filter(PubSubTopic.name==input.topic_name).\
                    one()

                callback = self._get_callback(session, input)

                sub_key = new_cid()
                consumer = PubSubConsumer(
                    None, input.is_active, sub_key, input.max_backlog, input.delivery_mode, callback[0],
                    callback[2], topic.id, input.client_id, input.cluster_id)

                session.add(consumer)
                session.commit()

            except Exception, e:
                msg = 'Could not create a consumer, e:`{}`'.format(format_exc(e))
                self.logger.error(msg)
                session.rollback()

                raise 
            else:

    def test_client(self):

        cid = new_cid()
        headers = {'x-zato-cid':cid}
        ok = True
        
        env = {
            'details': rand_string(),
            'result': ZATO_OK,
            'cid': cid
        }
        
        sio_payload_key = rand_string()
        sio_payload = {rand_string(): rand_string()}
        
        sio_response = {
            'zato_env': env,
            sio_payload_key: sio_payload
        }
        
        text = dumps(sio_response)
        status_code = rand_int()
        
        client = self.get_client(FakeInnerResponse(headers, ok, text, status_code))
        response = client.invoke()
        
        eq_(response.ok, ok)
        eq_(response.inner.text, text)
        eq_(response.data.items(), sio_response[sio_payload_key].items())
        eq_(response.has_data, True)
        eq_(response.cid, cid)
        eq_(response.cid, sio_response['zato_env']['cid'])
        eq_(response.details, sio_response['zato_env']['details'])

 def dispatcher_callback(self, event, ctx, **opaque):
     self.dispatcher_backlog.append(bunchify({
         'event_id': new_cid(),
         'event': event,
         'ctx': ctx,
         'opaque': opaque
     }))

 def test_repr(self):
     
     class MyResponse(_Response):
         def init(self):
             pass
     
     cid = new_cid()
     ok = True
     text = rand_string()
     status_code = rand_int()
     inner_params = ({'x-zato-cid':cid}, ok, text, status_code)
     
     max_repr = ((3,3), (len(text), CID_NO_CLIP))
     for(max_response_repr, max_cid_repr) in max_repr:
         
         inner = FakeInnerResponse(*inner_params)
         response = MyResponse(inner, False, max_response_repr, max_cid_repr, None)
         response.ok = ok
         
         cid_ellipsis = '' if max_cid_repr == CID_NO_CLIP else '..'
         
         expected = 'ok:[{}] inner.status_code:[{}] cid:[{}{}{}], inner.text:[{}]>'.format(
             ok, status_code, cid[:max_cid_repr], cid_ellipsis, cid[-max_cid_repr:], text[:max_response_repr])
         
         eq_(repr(response).endswith(expected), True)

    def get_data(self, data_format, transport, add_string=NON_ASCII_STRING, needs_payload=True,
            payload='', service_class=DummyAdminService):
        bmh = channel._BaseMessageHandler()
        
        expected = {
            'key': 'a' + uuid4().hex + add_string,
            'value': uuid4().hex + NON_ASCII_STRING,
            'result': uuid4().hex,
            'details': uuid4().hex,
            'cid': new_cid(),
        }
        
        if needs_payload:
            if not payload:
                if data_format == SIMPLE_IO.FORMAT.JSON:
                    payload_value = {expected['key']: expected['value']}
                else:
                    # str.format can't handle Unicode arguments http://bugs.python.org/issue7300
                    payload_value = '<%(key)s>%(value)s</%(key)s>' % (expected)
                payload = DummyPayload(payload_value)
        else:
            payload = None

        response = DummyResponse(payload, expected['result'], expected['details'])
        service = service_class(response, expected['cid'])

        bmh.set_payload(response, data_format, transport, service)
        
        return expected, service

    def test_invoke_retry_ok(self):

        target = 'target_{}'.format(rand_string())
        callback = 'callback_{}'.format(rand_string())
        callback_impl_name = 'callback_impl_name_{}'.format(rand_string())
        cid = new_cid()
        expected_result = rand_string()

        invoking_service = DummyTargetService(callback, callback_impl_name, cid, expected_result)
        ir = InvokeRetry(invoking_service)

        kwargs = {
            'async_fallback': True,
            'callback': callback,
            'context': {rand_string():rand_string()},
            'repeats': rand_int(),
            'seconds': rand_int(),
            'minutes': 0,
            'cid': cid,
        }

        result = ir.invoke(target, 1, 2, 3, **kwargs)
        self.assertEquals(expected_result, result)

        self.assertTrue(len(invoking_service.invoke_args), 2)
        self.assertEquals(invoking_service.invoke_args, (target, 1, 2, 3))
        self.assertEquals(invoking_service.invoke_kwargs, {'cid':cid})

 def __init__(self, kvdb, client_type, topic_callbacks, initial_lua_programs):
     self.kvdb = kvdb
     self.decrypt_func = kvdb.decrypt_func
     self.name = '{}-{}'.format(client_type, new_cid())
     self.topic_callbacks = topic_callbacks
     self.lua_container = LuaContainer(self.kvdb.conn, initial_lua_programs)
     self.ready = False

    def _invoke_callbacks(self, target, target_type, delivery, target_ok, in_doubt, invoker):
        """ Asynchronously notifies all callback services of the outcome of the target's invocation.
        """
        callback_list = delivery.definition.callback_list
        callback_list = callback_list.split(',') or []
        
        payload = dumps({
            'target_ok': target_ok,
            'in_doubt': in_doubt,
            'task_id': delivery.task_id,
            'target': target,
            'target_type': target_type,
            'invoker': invoker
        })

        for service in callback_list:
            if service:
                broker_msg = {}
                broker_msg['action'] = SERVICE.PUBLISH.value
                broker_msg['task_id'] = delivery.task_id
                broker_msg['channel'] = CHANNEL.DELIVERY
                broker_msg['data_format'] = DATA_FORMAT.JSON
                broker_msg['service'] = service
                broker_msg['payload'] = payload
                broker_msg['cid'] = new_cid()
                
                try:
                    self.broker_client.invoke_async(broker_msg)
                except Exception, e:
                    msg = 'Could not invoke callback:[%s], task_id:[%s], e:[%s]'.format(
                        service, delivery.task_id, format_exc(e))
                    self.logger.warn(msg)

    def setUp(self):
        self.key_prefix = 'zato:pubsub:{}:'.format(new_cid())
        self.kvdb = Redis()

        try:
            self.kvdb.ping()
        except ConnectionError:
            self.has_redis = False
        else:
            self.has_redis = True

 def invoke_async(self, msg, msg_type=MESSAGE_TYPE.TO_PARALLEL_ANY, expiration=BROKER.DEFAULT_EXPIRATION):
     msg['msg_type'] = msg_type
     msg = dumps(msg)
     
     topic = TOPICS[msg_type]
     key = broker_msg = b'zato:broker{}:{}'.format(KEYS[msg_type], new_cid())
     
     self.kvdb.conn.set(key, str(msg))
     self.kvdb.conn.expire(key, expiration)  # In seconds
     
     self.pub_client.publish(topic, broker_msg)

 def send(self, msg):
     msg['msg_type'] = MESSAGE_TYPE.TO_PARALLEL_ANY
     msg = dumps(msg)
     
     topic = TOPICS[MESSAGE_TYPE.TO_PARALLEL_ANY]
     key = broker_msg = b'zato:broker:to-parallel:any:{}'.format(new_cid())
     
     self.kvdb.conn.set(key, str(msg))
     self.kvdb.conn.expire(key, 15) # In seconds, TODO: Document it and make configurable
     
     self.pub_client.publish(topic, broker_msg)

 def deliver(self, def_name, payload, task_id=None, *args, **kwargs):
     """ Uses guaranteed delivery to send payload using a delivery definition known by def_name.
     *args and **kwargs will be passed directly as-is to the target behind the def_name.
     """
     task_id = task_id or new_cid()
     self.delivery_store.deliver(
         self.server.cluster_id, def_name, payload, task_id, self.invoke, 
         kwargs.pop('is_resubmit', False), 
         kwargs.pop('is_auto', False), 
         *args, **kwargs)
     
     return task_id

 def _on_message(self, msg, args):
     """ Invoked for each message taken off a ZMQ socket.
     """
     with self.channel_lock:
         params = {}
         params['action'] = CHANNEL.ZMQ_MESSAGE_RECEIVED
         params['service'] = self.channel.service
         params['cid'] = new_cid()
         params['payload'] = msg
         params['data_format'] = self.channel.data_format
         
         self.broker_client.invoke_async(params)

    def subscribe(self, ctx, sub_key=None):
        """ Subscribes the client to one or more topics, or topic patterns. Returns subscription key
        to use in subsequent calls to fetch messages by.
        """
        sub_key = sub_key or new_cid()
        with self.update_lock:
            for topic in ctx.topics:
                # TODO: Resolve topic here - it can be a pattern instead of a concrete name
                self.add_subscription(sub_key, ctx.client_id, topic)

        self.logger.info('Client `%s` sub to topics `%s`', ctx.client_id, ', '.join(ctx.topics))
        return sub_key