nakadi.py 文件源码

def value_callback(self, value):
        async_max_callback = None
        async_callback = None
        with self.guard:
            if value in self.async_callbacks:
                count, async_callback, async_max_callback = self.async_callbacks.pop(value)
                if count != self.receivers:
                    async_callback = None
                count -= 1
                if count > 0:
                    self.async_callbacks[value] = count, None, async_max_callback
                    async_max_callback = None
            else:
                logging.error('Callback for instance {} and value {} is not found'.format(self.instance_id, value))
        if async_callback:
            IOLoop.instance().add_callback(async_callback)
        if async_max_callback:
            IOLoop.instance().add_callback(async_max_callback)