strawhatfy/totoro

Celery integration with Tornado

totoro's Issues

Please upload to pypi with python3 support

The pypi's version does not support python3

Remove dependencies on redis and tornado-redis

Is it possible to remove the dependencies on redis and tornado-redis if Redis is not being used?

it doesn't work with celery4.0.2

The source code looks like this:

class MainHandler(web.RequestHandler):
    @gen.coroutine
    def post(self):
        response = yield gen.Task(tasks.test.apply_async, args=['abcdefghi'])
        self.write(response)

if __name__ == "__main__":
    totoro.setup_producer()
    agent = web.Application([(r'/', MainHandler),])
    agent.listen(8080)
    ioloop.IOLoop.instance().start()

I am using Redis as the broker and backend

$ python agent_totoro.py

Traceback (most recent call last):
  File "agent_totoro.py", line 37, in <module>
    totoro.setup_producer()
  File "/home/ryan/agent/totoro/__init__.py", line 60, in setup_producer
    _setup_producer(celery_app, io_loop, task_publish_delegate, result_cls)
  File "/home/ryan/agent/totoro/__init__.py", line 41, in _setup_producer
    producer_cls.register_consumer(RedisBackend, totoro.redis.RedisConsumer)
  File "/home/ryan/agent/totoro/base.py", line 105, in register_consumer
    raise ValueError('`backend_type` must be a subclass of BaseBackend.')
ValueError: `backend_type` must be a subclass of BaseBackend.

Comment the lines in register_consumer() an bypass this check, but still it throw error like below:

HTTPServerRequest(protocol='http', host='www.xxx.com:8080', method='POST', uri='/', version='HTTP/1.1', remote_ip='x.x.x.x', headers={'Content-Length': '266', 'Accept-Encoding': 'gzip, deflate', 'Connection': 'keep-alive', 'Accept': '*/*', 'User-Agent': 'python-requests/2.4.3 CPython/2.6.6 Linux/2.6.32-504.3.3.el6.x86_64', 'Host': 'www.xxx.com:8080'})
Traceback (most recent call last):
  File "/usr/lib64/python2.7/site-packages/tornado/web.py", line 1469, in _execute
    result = yield result
  File "/usr/lib64/python2.7/site-packages/tornado/gen.py", line 1016, in run
    value = future.result()
  File "/usr/lib64/python2.7/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "/usr/lib64/python2.7/site-packages/tornado/gen.py", line 1022, in run
    yielded = self.gen.throw(*exc_info)
  File "agent_totoro.py", line 33, in post
    response = yield gen.Task(tasks.test.apply_async, args=['abcdefghi'])
  File "/usr/lib64/python2.7/site-packages/tornado/gen.py", line 1016, in run
    value = future.result()
  File "/usr/lib64/python2.7/site-packages/tornado/concurrent.py", line 237, in result
    raise_exc_info(self._exc_info)
  File "/usr/lib64/python2.7/site-packages/tornado/gen.py", line 588, in Task
    func(*args, callback=_argument_adapter(set_result), **kwargs)
  File "/usr/lib/python2.7/site-packages/celery/app/task.py", line 535, in apply_async
    **options
  File "/usr/lib/python2.7/site-packages/celery/app/base.py", line 737, in send_task
    amqp.send_task_message(P, name, message, **options)
  File "/usr/lib/python2.7/site-packages/celery/app/amqp.py", line 558, in send_task_message
    **properties
  File "/usr/lib/python2.7/site-packages/kombu/messaging.py", line 181, in publish
    exchange_name, declare,
  File "/usr/lib/python2.7/site-packages/kombu/connection.py", line 494, in _ensured
    return fun(*args, **kwargs)
  File "/usr/lib/python2.7/site-packages/kombu/messaging.py", line 203, in _publish
    mandatory=mandatory, immediate=immediate,
  File "/usr/lib/python2.7/site-packages/kombu/transport/virtual/base.py", line 608, in basic_publish
    return self._put(routing_key, message, **kwargs)
  File "/usr/lib/python2.7/site-packages/kombu/transport/redis.py", line 767, in _put
    client.lpush(self._q_for_pri(queue, pri), dumps(message))
  File "/usr/lib/python2.7/site-packages/kombu/utils/json.py", line 72, in dumps
    **dict(default_kwargs, **kwargs))
  File "/usr/lib64/python2.7/json/__init__.py", line 250, in dumps
    sort_keys=sort_keys, **kw).encode(obj)
  File "/usr/lib64/python2.7/json/encoder.py", line 207, in encode
    chunks = self.iterencode(o, _one_shot=True)
  File "/usr/lib64/python2.7/json/encoder.py", line 270, in iterencode
    return _iterencode(o, 0)
  File "/usr/lib/python2.7/site-packages/kombu/utils/json.py", line 62, in default
    return super(JSONEncoder, self).default(o)
  File "/usr/lib64/python2.7/json/encoder.py", line 184, in default
    raise TypeError(repr(o) + " is not JSON serializable")
TypeError: <function wrapper at 0x2e4cc80> is not JSON serializable