malinoff / amqproto Goto Github PK

View Code? Open in Web Editor NEW

22.0 22.0 2.0 318 KB

License: Apache License 2.0

Python 100.00%

amqproto's People

Contributors

Stargazers

Watchers

Forkers

wasp singulared

amqproto's Issues

Hanging on sending message with str body via basic publush

Steps to reproduce:

git clone current master @ 7081528
pip install -e .[all]
Run this script

import asyncio
import json
from amqproto.io.asyncio import Connection
from amqproto.protocol import BasicMessage

async def main():
    async with Connection(host='127.0.0.1') as connection:
        async with connection.get_channel() as channel:
            message = BasicMessage(json.dumps({'ok': True}))
            await channel.basic_publish(message, routing_key='hello')

loop = asyncio.get_event_loop()
loop.run_until_complete(main())

ctrl+C

$ python test.py 
^CTraceback (most recent call last):
  File "test.py", line 15, in <module>
    loop.run_until_complete(main())
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 454, in run_until_complete
    self.run_forever()
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 421, in run_forever
    self._run_once()
  File "/usr/lib64/python3.6/asyncio/base_events.py", line 1389, in _run_once
    event_list = self._selector.select(timeout)
  File "/usr/lib64/python3.6/selectors.py", line 445, in select
    fd_event_list = self._epoll.poll(timeout, max_ev)
KeyboardInterrupt
Exception ignored in: <coroutine object main at 0x7f07d348a8e0>
RuntimeError: coroutine ignored GeneratorExit
Task exception was never retrieved
future: <Task finished coro=<Connection._communicate() done, defined at amqproto/amqproto/io/asyncio.py:294> exception=<AMQPError: 505, 60, 40>>
Traceback (most recent call last):
  File "amqproto/amqproto/io/asyncio.py", line 307, in _communicate
    coroutine = handler(frame)
  File "amqproto/amqproto/connection.py", line 162, in handle_frame
    return handler(method)
  File "amqproto/amqproto/connection.py", line 364, in _receive_ConnectionClose
    return self._send_ConnectionCloseOK(exc)
  File "amqproto/amqproto/connection.py", line 369, in _send_ConnectionCloseOK
    raise _exc
amqproto.protocol.errors.UnexpectedFrame: UNEXPECTED_FRAME - expected content body, got non content body frame instead

While the fix is to encode body into bytes, the error is quite misleading.

Massive queues deletion eventually hangs

Case: delete 130K queues. Script:

import asyncio
from amqproto.io.asyncio import Channel, Connection


async def main():
    with open('queues.txt') as fobj:
        queues = fobj.read().splitlines()
    connection = Connection()
    async with connection:  # type: Connection
        async with connection.get_channel() as channel:  # type: Channel
            for idx, queue in enumerate(queues):
                await channel.queue_delete(queue)
                print(idx)

asyncio.get_event_loop().run_until_complete(main())

May be not optimal, but what I made quickly. Counter keeps printing quite fast until some moment (100-200 items) and then script hangs and nothing happens more than for a minute. Rerun causes the same.

Meanwhile, pika deletes queues much slower, also eventually hangs, but for short periods, about dozen seconds (I suspect due to RabbitMQ GC or else stuff) and then keeps going.

It also could reproduced with less number of queues: for 30K amqproto successfully deleted 12K queues before hang.

P.S. amqproto @ 7081528

Implement curio adapter

Doesn't worth it until several testing related issues are fixed:

Fatal Python error: GC object already tracked, pytest-dev/pytest-asyncio#61
pytest-curio should be able to work with the newest curio, johnnoone/pytest-curio#1
pytest-curio should be able to handle async (generator) fixtures, johnnoone/pytest-curio#4

attrs version conflict

Steps to reproduce:

git clone current master @ 7081528
pip install -e .[all]
py.test

Traceback (most recent call last):
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 510, in load_setuptools_entrypoints
    plugin = ep.load()
  File "venv/lib64/python3.6/site-packages/pkg_resources/__init__.py", line 2290, in load
    self.require(*args, **kwargs)
  File "venv/lib64/python3.6/site-packages/pkg_resources/__init__.py", line 2307, in require
    items = working_set.resolve(reqs, env, installer)
  File "venv/lib64/python3.6/site-packages/pkg_resources/__init__.py", line 858, in resolve
    raise VersionConflict(dist, req).with_context(dependent_req)
pkg_resources.VersionConflict: (attrs 17.2.0 (venv/lib/python3.6/site-packages), Requirement.parse('attrs<17,>=16'))

During handling of the above exception, another exception occurred:

Traceback (most recent call last):
  File "venv/bin/py.test", line 11, in <module>
    sys.exit(main())
  File "venv/lib64/python3.6/site-packages/_pytest/config.py", line 49, in main
    config = _prepareconfig(args, plugins)
  File "venv/lib64/python3.6/site-packages/_pytest/config.py", line 168, in _prepareconfig
    pluginmanager=pluginmanager, args=args)
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 745, in __call__
    return self._hookexec(self, self._nonwrappers + self._wrappers, kwargs)
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 339, in _hookexec
    return self._inner_hookexec(hook, methods, kwargs)
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 334, in <lambda>
    _MultiCall(methods, kwargs, hook.spec_opts).execute()
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 613, in execute
    return _wrapped_call(hook_impl.function(*args), self.execute)
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 250, in _wrapped_call
    wrap_controller.send(call_outcome)
  File "venv/lib64/python3.6/site-packages/_pytest/helpconfig.py", line 68, in pytest_cmdline_parse
    config = outcome.get_result()
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 279, in get_result
    raise ex[1].with_traceback(ex[2])
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 265, in __init__
    self.result = func()
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 614, in execute
    res = hook_impl.function(*args)
  File "venv/lib64/python3.6/site-packages/_pytest/config.py", line 957, in pytest_cmdline_parse
    self.parse(args)
  File "venv/lib64/python3.6/site-packages/_pytest/config.py", line 1121, in parse
    self._preparse(args, addopts=addopts)
  File "venv/lib64/python3.6/site-packages/_pytest/config.py", line 1084, in _preparse
    self.pluginmanager.load_setuptools_entrypoints('pytest11')
  File "venv/lib64/python3.6/site-packages/_pytest/vendored_packages/pluggy.py", line 515, in load_setuptools_entrypoints
    "Plugin %r could not be loaded: %s!" % (ep.name, e))
_pytest.vendored_packages.pluggy.PluginValidationError: Plugin 'docker' could not be loaded: (attrs 17.2.0 (venv/lib/python3.6/site-packages), Requirement.parse('attrs<17,>=16'))!

Can we avoid using Futures in the core?

Currently Future objects are in the core to provide a number of very convenient behaviors:

It is a synchronization mechanism: AMQP methods that have replies simply own a Future instance that is being set in reply handlers when replies arrive.
It is a callback-based API replacement: the caller has interest in responding to certain events, like handling an incoming message after channel.basic_consume, and instead of providing a callback to basic_consume the caller simply awaits the returned Future (the exact method of awaiting depends on the actual I/O implementation).
It is a way to have a very concise and boilerplate-less I/O adapters implementations: for instance, amqproto.Channel.basic_consume method cannot be defined as async method (for obvious reasons). If basic_consume does not return something that can be awaited, then we would need to basically re-define all (34) AMQP methods in the asyncio adapter and either mark them as async methods or return awaitables from them. Which is not appropriate, in my opinion.

Is there a way to keep all three conveniences and remove the use of Future objects in the core?

malinoff / amqproto Goto Github PK

amqproto's People

Contributors

Stargazers

Watchers

Forkers

amqproto's Issues

Hanging on sending message with str body via basic publush

Massive queues deletion eventually hangs

Implement curio adapter

attrs version conflict

Can we avoid using Futures in the core?

Recommend Projects

React

Vue.js

Typescript

TensorFlow

Django

Laravel

D3

Recommend Topics

javascript

web

server

Machine learning

Visualization

Game

Recommend Org

Facebook

Microsoft

Google

Alibaba

D3

Tencent