smok-serwis / coolamqp Goto Github PK

View Code? Open in Web Editor NEW

6.0 2.0 0.0 957 KB

Pure Python pretty fast AMQP client with sane reconnects

License: MIT License

Python 99.71% Shell 0.17% Dockerfile 0.12%

amqp-client pypi rabbitmq consumer amqp amqp0-9-1

coolamqp's People

Contributors

Stargazers

Watchers

coolamqp's Issues

What do you do when you're forking?

Needs to support no_ack in consume

A battlefield report

Usługa zgłaszająca: svcr@dev8
Wersja smok4/core: 4.5.27
Czas zgłoszenia: 2019-12-25 01:17:55 czasu lokalnego
SMOK4_SLOT=8
SMOK4_SERVICE=svcr
Etykieta zdarzenia: handler.failed
Klasa zdarzenia: develop/fail

trigger : HandlerStopOrder(device_id='101187585uart')

trackback : Traceback (most recent call last):
  File "/usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py", line 69, in handle_msg
    self.inner_handle_msg(msg, msgtype)
  File "/usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py", line 96, in inner_handle_msg
    self.__intercept_unload(msg.device_id, None, None, None)
  File "/usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py", line 251, in __intercept_unload
    self.vport_data_mapper.device_shutdown(device_id)
  File "/usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/trackers.py", line 131, in device_shutdown
    self.conmap[(k[0], k[1], device_id)].cancel()
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/consumer.py", line 191, in cancel
    self.on_close)
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/channeler.py", line 197, in method_and_watch
    assert self.channel_id is not None
AssertionError

* Stack trace, innermost first
** method_and_watch at /usr/local/lib/python2.7/dist-packages/coolamqp/attaches/channeler.py:197
*** callable: <bound method Consumer.on_close of <coolamqp.attaches.consumer.Consumer object at 0x7f89b4403890>>
*** self: <coolamqp.attaches.consumer.Consumer object at 0x7f89b4403890>
*** method_classes_to_watch: [<class 'coolamqp.framing.definitions.BasicCancelOk'>]
*** method_payload: BasicCancel('12f688eb37c9477ab003307dc35750ea', False)
** cancel at /usr/local/lib/python2.7/dist-packages/coolamqp/attaches/consumer.py:191
*** self: <coolamqp.attaches.consumer.Consumer object at 0x7f89b4403890>
** device_shutdown at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/trackers.py:131
*** x: {(119275521, 0): ('119275521uart', <function wrap_data_handler at 0x7f89b4469848>), (109641729, 1): ('109641729gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b4032250>>), (105185281, 1): ('105185281gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c049cbd0>>), (112984065, 0): ('112984065uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89c04a16d0>>), (109314049, 1): ('109314049gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89944bae10>>), (101187585, 0): ('101187585uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b44ba550>>), (100073473, 0): ('100073473uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b44df450>>), (114884609, 0): ('114884609', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b42def50>>), (113049601, 2): ('113049601at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b4522210>>), (13959169, 1): ('13959169gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b4076b10>>), (109051905, 0): ('109051905uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b4060590>>), (105185281, 2): ('105185281at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89c047e3d0>>), (113836033, 1): ('113836033gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c052a810>>), (121438209, 0): ('121438209uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f8994455b10>>), (101974017, 1): ('101974017gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b43b6910>>), (123207681, 0): ('114556929uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b42decd0>>), (103677953, 1): ('103677953gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c0518510>>), (109707265, 1): ('109707265gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b4047a90>>), (106823681, 2): ('106823681at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89c0523a10>>), (101580801, 2): ('101580801at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b43af250>>), (118947841, 2): ('118947841at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b45146d0>>), (107806721, 2): ('107806721at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b40787d0>>), (107282433, 1): ('107282433gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c049b8d0>>), (118489089, 0): ('118489089uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89c0491f50>>), (720897, 1): ('1', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b4403b90>>), (15007745, 1): ('15007745gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f899447b410>>), (105709569, 0): ('105709569uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89c0518150>>), (102563841, 2): ('102563841at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b45287d0>>), (103677953, 2): ('103677953at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89c04b86d0>>), (15925249, 0): ('15925249', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f899447b510>>), (113836033, 0): ('113836033uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89c0491110>>), (108331009, 0): ('108331009uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b409b6d0>>), (101449729, 0): ('101449729uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b43ae050>>), (102957057, 1): ('102957057gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c05232d0>>), (104529921, 1): ('104529921gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c04d5890>>), (101777409, 1): ('101777409gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b43e2a50>>), (15597569, 2): ('15597569at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89944e4dd0>>), (123207681, 1): ('114556929gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c0519f50>>), (105644033, 0): ('105644033uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89c04f3950>>), (119275521, 2): ('119275521at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89944d9590>>), (108331009, 1): ('108331009gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b408bc90>>), (119013377, 2): ('119013377at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b40a3f90>>), (109641729, 2): ('109641729at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b4047f90>>), (111673345, 2): ('111673345at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89b44c68d0>>), (14811137, 1): ('14811137gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89c0519850>>), (10878977, 1): ('10878977', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b40a1d50>>), (109314049, 0): ('109314049uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89944c7150>>), (15007745, 2): ('15007745at', <bound method SVCRHandler.on_sssp_data of <svcr.apps.csqtrack.SVCRHandler object at 0x7f89944c74d0>>), (101318657, 0): ('101318657uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b43ba290>>), (120848385, 1): ('120848385gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89b4085290>>), (119341057, 1): ('119341057gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f89944ea3d0>>), (121438209, 1): ('121438209gpio', <bound method SVCRHandler.on_sssp_data of <svcr.apps.gpio.SVCRHandler object at 0x7f899449ed50>>)}
*** k: (101187585, 0)
*** device_id: '101187585uart'
*** self: <svcr.worker.eventloop.trackers.VportDataMapper object at 0x7f89b81b0150>
*** v: ('101187585uart', <bound method SVCRHandler.on_sssp_data of <svcr.apps.semanta.SVCRHandler object at 0x7f89b44ba550>>)
** __intercept_unload at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py:251
*** exception: None
*** report_to_catalog: False
*** self: <EventLoop(eventloop, started 140229448578816)>
*** trigger: None
*** trackback: None
*** device_id: '101187585uart'
** inner_handle_msg at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py:96
*** msg: HandlerStopOrder(device_id='101187585uart')
*** self: <EventLoop(eventloop, started 140229448578816)>
*** msgtype: <class 'svcr.worker.eventloop.tqmifc.HandlerStopOrder'>
** handle_msg at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py:86
*** e: AssertionError()
*** self: <EventLoop(eventloop, started 140229448578816)>
*** msgtype: <class 'svcr.worker.eventloop.tqmifc.HandlerStopOrder'>
*** msg: HandlerStopOrder(device_id='101187585uart')
*** has_device_id: True
*** device_id: '101187585uart'
** inner_run at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/worker/eventloop/__init__.py:205
*** msg: HandlerStopOrder(device_id='101187585uart')
*** self: <EventLoop(eventloop, started 140229448578816)>
*** msgs: <sai.moves.satella1.IterableReader object at 0x7f89b81b01d0>
** run at /usr/local/lib/python2.7/dist-packages/smok4svcr-4.5.18rc1-py2.7.egg/svcr/utils.py:46
*** self: <EventLoop(eventloop, started 140229448578816)>
** run_with_except_hook at /usr/local/lib/python2.7/dist-packages/smok4core-4.5.27-py2.7.egg/sai/basic/__init__.py:90
*** args: ()
*** kw: {}
*** run_old: <bound method EventLoop.run of <EventLoop(eventloop, started 140229448578816)>>
** __bootstrap_inner at /usr/lib/python2.7/threading.py:801
*** self: <EventLoop(eventloop, started 140229448578816)>
** __bootstrap at /usr/lib/python2.7/threading.py:774
*** self: <EventLoop(eventloop, started 140229448578816)>

Use librabbitmq if available

This, according to guys from Celery, should be a drop-in replacement, but suddenly all tests fail :D

Support passing qos as int

it should be understood as None, int then.

Consumer is not marked as cancelled when it gets a CCN

CoolAMQP hangs if server is unavailable upon connect

It should've at least thrown an error

A non-hashed MessageProperties

Right now MessageProperties will hash entire key=>values and store it within non-GCable memory in order to facilitate speed and memory consumption.

Some people want to ship each message with different properties, and they need a workaround for that.

Add some way to notify user of consumer cancel

RMQ extension

Weird AssertionErrors

See Travis CI, latest jobs that failed. Or just any jobs.

Why, AssertionError?

Consuming an exclusive queue kills the coolamqp thread

Add fallback for TX if publisher confirms are not available

use standard AMQP transactions for achieving transaction safety on messages published with tx=True

Consuming twice from a queue kills the thread

Thread reports back with something like:

Jun 02 15:47:50 dev1 pypy[11422]: return self.on_inbound_frame(read_frame())
Jun 02 15:47:50 dev1 pypy[11422]: File "/usr/local/lib/pypy2.7/dist-packages/amqp/method_framing.py", line 63, in on_frame
Jun 02 15:47:50 dev1 pypy[11422]: callback(channel, method_sig, buf, None)
Jun 02 15:47:50 dev1 pypy[11422]: File "/usr/local/lib/pypy2.7/dist-packages/amqp/connection.py", line 418, in on_inbound_method
Jun 02 15:47:50 dev1 pypy[11422]: method_sig, payload, content,
Jun 02 15:47:50 dev1 pypy[11422]: File "/usr/local/lib/pypy2.7/dist-packages/amqp/abstract_channel.py", line 138, in dispatch_method
Jun 02 15:47:50 dev1 pypy[11422]: listener(*args)
Jun 02 15:47:50 dev1 pypy[11422]: File "/usr/local/lib/pypy2.7/dist-packages/amqp/connection.py", line 543, in _on_close
Jun 02 15:47:50 dev1 pypy[11422]: (class_id, method_id), ConnectionError)
Jun 02 15:47:50 dev1 pypy[11422]: NotAllowed: Basic.consume: (530) NOT_ALLOWED - attempt to reuse consumer tag 'xxxx'```

Encoding of messages

Message bodies are bunch of bytes. They should be returned that way too

coolamqp.objects.Callable is not threadsafe

And most of the time it doesn't need to be.

A new thread-safe class should be made not to hurt performance in existing places.

https://travis-ci.org/smok-serwis/coolamqp/jobs/190760684

wanted - dead or alive

Bug spotted

----------------------------------------------------------------------
Traceback (most recent call last):
  File "/builds/smok4/core/tests/test_shci/test_drivers/test_otbox/test_drvobj.py", line 17, in test_get_parameters_curacao
    dev = create_nothingdevice(please='shci/otbox')
  File "/builds/smok4/core/tests/utils.py", line 49, in create_nothingdevice
    install(device_id, True, 'flow')
  File "/builds/smok4/core/shci/drivers/otbox/install.py", line 170, in install
    Predicate.create(device_id, 'opentherm', 'OT', {}).start()
  File "/builds/smok4/core/sai/alarms/predicate.py", line 153, in create
    start.run(device_id, new_id)
  File "/usr/local/lib/python2.7/dist-packages/loveseal/task.py", line 163, in run
    self.region.amqp.publish(msg, routing_key=self.region.deadletter_queue, tx=True).result()
  File "/builds/smok4/core/sai/basic/msgs.py", line 83, in publish
    return super(_AMQP, self).publish(msg, exchange, routing_key, tx=tx)
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/clustering/cluster.py", line 144, in publish
    return (self.pub_tr if tx else self.pub_na).publish(message, exchange, routing_key)
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/utils.py", line 231, in monitored
    return fun(*args, **kwargs)
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/publisher.py", line 218, in publish
    self._mode_cnpub_process_deliveries()
  File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/publisher.py", line 157, in _mode_cnpub_process_deliveries
    self.tagger.deposit(self.tagger.get_key(), FutureConfirmableRejectable(fut))
AttributeError: 'NoneType' object has no attribute 'deposit'
-------------------- >> begin captured logging << --------------------
coolamqp.uplink.connection.connection: DEBUG: TCP connection established, authentication in progress
sai.basic.msgs: INFO: Finished AMQP setup
coolamqp.attaches.publisher: DEBUG: Publish request, but not connected - dropping the message
coolamqp.uplink.connection.connection: DEBUG: Received connection.start
coolamqp.uplink.connection.connection: DEBUG: Received connection.tune
coolamqp.uplink.connection.connection: DEBUG: Received connection.open-ok
coolamqp.uplink.connection.connection: INFO: Connection ready.
coolamqp.uplink.connection.connection: DEBUG: Received channel.open-ok
coolamqp.uplink.connection.connection: DEBUG: Received channel.open-ok
coolamqp.attaches.publisher: DEBUG: Publisher on_setup, payload=<coolamqp.framing.definitions.ChannelOpenOk object at 0x7ff79c5c1410>
coolamqp.attaches.publisher: INFO: Publisher noack is up
coolamqp.uplink.connection.connection: DEBUG: Received channel.open-ok
coolamqp.attaches.publisher: DEBUG: Publisher on_setup, payload=<coolamqp.framing.definitions.ChannelOpenOk object at 0x7ff79c5c1410>
coolamqp.uplink.connection.connection: DEBUG: Received confirm.select-ok
coolamqp.attaches.publisher: DEBUG: Publisher on_setup, payload=<coolamqp.framing.definitions.ConfirmSelectOk object at 0x7ff79c536b10>
--------------------- >> end captured logging << ---------------------

Add capability to add sth to client properties

Problem with binding anonymous queues to topic exchanges

This feature is broken. Create your own queues that are auto_delete and have unique names for now.

Add support for gevent

So far, CoolAMQP uses an epoll-based approach, which doesn't work for greenlets. Write a select-based listener.

Cluster.publish - tx kwarg is confusing

because it doesn't force transactionality, but only a confirm.

Traceback (most recent call last):
File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner
self.run()
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/listener/thread.py", line 27, in run
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/listener/epoll_listener.py", line 86, in wait
sock.on_read()
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/listener/socket.py", line 98, in on_read
self.my_on_read(data)
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/connection/recv_framer.py", line 60, in put
while self._statemachine():
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/connection/recv_framer.py", line 140, in _statemachine
self.on_frame(frame)
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/connection/connection.py", line 217, in on_frame
watch_triggered = watch.is_triggered_by(frame)
File "/usr/local/lib/python2.7/dist-packages/coolamqp/uplink/connection/watches.py", line 126, in is_triggered_by
self.callable(frame.payload)
File "/usr/local/lib/python2.7/dist-packages/coolamqp/attaches/consumer.py", line 234, in on_delivery
self.receiver.on_basic_deliver(sth)
AttributeError: 'NoneType' object has no attribute 'on_basic_deliver'

        try:
            mode = message.properties.content_type.tobytes()
        except AttributeError:
            mode = b'application/x-pickle'

all over sucks

Add support for channel flow control