strawhatfy/totoro

Celery integration with Tornado

totoro's Issues

Please upload to pypi with python3 support

The pypi's version does not support python3

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

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?