Thread-safe Python RabbitMQ Client & Management library

Whoooo, more wierdness!

.Thread-21 - INFO - Message channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message properties: {'headers': None, 'reply_to': '', 'user_id': '', 'cluster_id': '', 'app_id': '', 'delivery_mode': None, 'content_type': '', 'correlation_id': 'keepalive', 'expiration': '', 'message_id': '', 'message_type': '', 'priority': None, 'content_encoding': 'utf-8', 'timestamp': None}
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _method: {'routing_key': b'nak', 'redelivered': False, 'delivery_tag': 5, 'exchange': b'keepalive_exchange140037604406920', 'consumer_tag': b'amq.ctag-DCdsyFR66UJsU7sgkNp9oQ'}
ACK For delivery tag: 5
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Heartbeat packet received! wat
Main.Connector.Internal(/rpcsys).Thread-23 - INFO - Timeout watcher loop. Current message counts: 0 (out: 0, in: 0)
Main.Connector.Container(/rpcsys).Thread-23 - INFO - Interface timeout thread. Ages: heartbeat -> 4.83, last message -> 32.51.
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Received message!
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message properties: {'headers': None, 'reply_to': '', 'user_id': '', 'cluster_id': '', 'app_id': '', 'delivery_mode': None, 'content_type': '', 'correlation_id': 'keepalive', 'expiration': '', 'message_id': '', 'message_type': '', 'priority': None, 'content_encoding': 'utf-8', 'timestamp': None}
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _method: {'routing_key': b'nak', 'redelivered': False, 'delivery_tag': 6, 'exchange': b'keepalive_exchange140037604406920', 'consumer_tag': b'amq.ctag-DCdsyFR66UJsU7sgkNp9oQ'}
ACK For delivery tag: 6
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Heartbeat packet received! wat
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Received message!
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message properties: {'headers': None, 'reply_to': '', 'user_id': '', 'cluster_id': '', 'app_id': '', 'delivery_mode': None, 'content_type': '', 'correlation_id': 'keepalive', 'expiration': '', 'message_id': '', 'message_type': '', 'priority': None, 'content_encoding': 'utf-8', 'timestamp': None}
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _channel: <amqpstorm.channel.Channel object at 0x7f5d08cdbcc8>
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Message _method: {'routing_key': b'nak', 'redelivered': True, 'delivery_tag': 7, 'exchange': b'keepalive_exchange140037604406920', 'consumer_tag': b'amq.ctag-DCdsyFR66UJsU7sgkNp9oQ'}
ACK For delivery tag: 7
Main.Connector.Container(/rpcsys).Thread-21 - INFO - Heartbeat packet received! wat
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR - Error while in rx runloop!
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -       Channel 1 was closed by remote server: PRECONDITION_FAILED - unknown delivery tag 6
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR - Traceback (most recent call last):
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -   File "/media/Storage/Scripts/ReadableWebProxy/FetchAgent/AmqpConnector/__init__.py", line 546, in _rx_poll
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -     self.interface.process_rx_events()
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -   File "/media/Storage/Scripts/ReadableWebProxy/FetchAgent/AmqpConnector/__init__.py", line 171, in process_rx_events
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -     self.storm_channel.process_data_events(to_tuple=False)
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -   File "/media/Storage/Scripts/ReadableWebProxy/amqpstorm/channel.py", line 266, in process_data_events
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -     for message in self.build_inbound_messages(break_on_empty=True):
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -   File "/media/Storage/Scripts/ReadableWebProxy/amqpstorm/channel.py", line 113, in build_inbound_messages
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -     self.check_for_errors()
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -   File "/media/Storage/Scripts/ReadableWebProxy/amqpstorm/channel.py", line 188, in check_for_errors
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -     raise exception
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR - amqpstorm.exception.AMQPChannelError: Channel 1 was closed by remote server: PRECONDITION_FAILED - unknown delivery tag 6
Main.Connector.Internal(/rpcsys).Thread-21 - ERROR -
Main.Connector.Internal(/rpcsys).Thread-21 - WARNING - Should die flag! Runstate: running, threads live: threads alive, had exception: yes.
Main.Connector.Internal(/rpcsys).Thread-21 - INFO - RX Poll process dying. Threads_live: 1, had exception 1, should_die True
Main.Connector.Internal(/rpcsys).Thread-22 - WARNING - Should die flag! Runstate: running, threads live: threads alive, had exception: yes.
Main.Connector.Internal(/rpcsys).Thread-22 - WARNING - Should die flag! Runstate: running, threads live: threads alive, had exception: yes.
Main.Connector.Internal(/rpcsys).Thread-22 - INFO - TX Poll process dying (should die: True). Threads_live: 1, runstate: 1, resp queue size: 0, had exception 1.
Main.Connector.Internal(/rpcsys).Thread-23 - WARNING - Should die flag! Runstate: running, threads live: threads alive, had exception: yes.
Disconnecting!
Main.Connector.Container(/rpcsys).Thread-2 - INFO - Closing channel
Main.Connector.Container(/rpcsys).Thread-2 - INFO - Closing connection
Running state: True, thread alive: True, thread id:140037440014080
Joining _inbound_thread. Runstate: %s False

Context:

I have a connection with a thread processing the receiving messages. I have instrumented Message.ack() with a print statement that prints the delivery tag that it's acking.

It appears I'm calling [delivery tag 6].ack(), [delivery tag 7].ack(), and somehow the ack for delivery tag 7 is getting received by the rabbitmq server /first/, resulting in a PRECONDITION_FAILED error because acking 7 implicitly acks previous tags, and therefore 6 is not a valid delivery tag anymore.

I'm working on pulling out a testable example, but it's certainly odd.

Incidentally, the new docs pages are fancypants!

Basically, I'm dealing with a context where I have a high volume traffic AMQP connection, across a high-latency, unreliable link (intercontinental).

If there is a way for a connection to possibly wedge, I'll encounter it.

Anyways, The issue I ran into here is that it's possible for the "close" operation on a connection to wedge indefinitely, preventing a connection from actually closing (I /think/ if the shutdown RPC request gets garbled/eaten somewhere).

The question of /how/ this happens aside, I therefore need to be able to kill a open connection in a dirty manner. This adds the ability to kill() a connection which will force the worker thread to exit immediately, without bothering to do any proper cleanup.

Because this is a operation that's generally done with a additional watcher process, outside of the normal program flow, I used multiprocessing primitives to manage the _die flag (at one point, the watcher was a separate /process/, not thread).

Anyways, I'm not sure if this is inline with the ideas of the library, but I haven't been able to wedge the amqp connection with this patch set, so that's something.

Only tested on Py3.5x64, Ubuntu 14.04.

Things I haven't done: Additional unit tests.

I regularly get [SSL: BAD_WRITE_RETRY] bad write retry (_ssl.c:1647) errors when running AMQP-Storm using SSL.

Looking at the code I think its because it's not handling SSL Retry exceptions from the socket

http://stackoverflow.com/questions/2997218/why-am-i-getting-error1409f07fssl-routinesssl3-write-pending-bad-write-retr

i'll get a stack trace to see if its on write or on do_handshake

Hello,

I've run into an issue with creating new channels and receiving the reached the maximum number of channels 65535 when attempting to create a new channel.

After some digging, I noticed Connection._get_next_available_channel_id() accounts for all channels, both open and closed. I believe filtering the count for just opened should resolve this issue.

I tested with a quick fix

    def _get_next_available_channel_id(self):
        channel_id = len(self._channels) + 1
        active_channels = [
            ch for ch in list(self._channels.values()) if ch and ch.is_open
        ]
        if len(active_channels) >= self.max_allowed_channels:
            raise AMQPConnectionError(
                'reached the maximum number of channels %d' %
                self.max_allowed_channels)
        return channel_id

However it may be better to just keep an active count

Upgraded to RabbitMq 3.6.2 starting to see the following errors which I think lead to using all the memory on rabbitmq server.

[Channel%d] Unhandled Frame: %s -- %s

No code changes, just a server upgrade. Not sure if this storm or rabbitmq.

I have been experiencing a strange error, just for some messages I got this error:

'Deliver' object has no attribute 'body_size' channel.py (line: 253)

I looked at self._inbound and I saw some Deliver objects in sequence instead of Deliver/ContentHeader/ContentBody

The messages are being sent to the queue by another project in another language, so I have no clue why it is happening.

This is a weird one, I've upgraded to 2.2 and trying to do the following script

import os
import logging
logging.basicConfig(level=logging.DEBUG)

from amqpstorm import Message
from amqpstorm import UriConnection

keys = ["1","2","3"]
#signer = Signer()
bus = UriConnection("***")
with bus.channel(rpc_timeout=10) as channel:
    channel.confirm_deliveries()
    for key in keys:
        print key
        msg = "My Message"
        #properties = {"headers": {"md5-signature": signer.sign(msg)}}
        Message.create(channel, msg, properties).publish(key, exchange="amq.topic")

I get this error on the second publish.

DEBUG:amqpstorm.connection:Connection Opening DEBUG:amqpstorm.channel0:Frame Received: Connection.Start DEBUG:amqpstorm.channel0:Frame Sent: Connection.StartOk DEBUG:amqpstorm.channel0:Frame Received: Connection.Tune DEBUG:amqpstorm.channel0:Frame Sent: Connection.TuneOk DEBUG:amqpstorm.channel0:Frame Sent: Connection.Open DEBUG:amqpstorm.channel0:Frame Received: Connection.OpenOk DEBUG:amqpstorm.heartbeat:Heartbeat Checker Started DEBUG:amqpstorm.connection:Connection Opened DEBUG:amqpstorm.connection:Opening a new Channel DEBUG:amqpstorm.connection:Channel #1 Opened 1 2 DEBUG:amqpstorm.channel:Channel #1 Closing DEBUG:amqpstorm.channel:Channel #1 Closed Exception in thread amqpstorm.io: Traceback (most recent call last): File "/usr/lib/python2.7/threading.py", line 801, in __bootstrap_inner self.run() File "/usr/lib/python2.7/threading.py", line 754, in run self.__target(*self.__args, **self.__kwargs) File "/home/adam/.venv/local/lib/python2.7/site-packages/amqpstorm/io.py", line 219, in _process_incoming_data if self.poller.is_ready: File "/home/adam/.venv/local/lib/python2.7/site-packages/amqpstorm/io.py", line 48, in is_ready except select.error as why: AttributeError: 'NoneType' object has no attribute 'error'

I plan on releasing pamqp 3.0 soon and wanted to make sure you were aware of the changes that are making it into 3.0 and give you an opportunity to report any issues prior to its release.

Please see the version history @ https://pamqp.readthedocs.io/en/latest/history.html

Hello,

I am using your library in conjunction with celery and i regularly encounter a problem: When asking celery to restart its worker (emitting a SIGTERM), the process is blocked and celery doesn't want to restart. Celery maintains a parent thread that runs child threads where tasks are executed. Amqpstorm's thread is also run from this parent thread. The processes in htop look like this:

Here process 13880 & 13803 are related to amqpstorm. 13803 is the inbound_thread while 13880 is the heartbeat timer.

After some investigation, i found out that killing the thread responsible for the heartbeat timer allows celery to gracefuly restart... This lead me to think that the timer could possibly create a deadlock.

I have created a pull request based on this assumption and will try this branch on my setup: Can you tell me what you think of it ? Do you see any other possible explanations for the error i see ?

Hello,

We recently updated the library version to 2.1.3 (from 1.1.7) and we now face several errors we did not have previously. One of them is the following:

    self.channel.basic.publish(body=data, exchange=self.exchange, routing_key='')
  File "/home/wirebot/.virtualenvs/cayzn_yield.etl/local/lib/python2.7/site-packages/amqpstorm/basic.py", line 194, in publish
    self._channel.write_frames(frames_out)
  File "/home/wirebot/.virtualenvs/cayzn_yield.etl/local/lib/python2.7/site-packages/amqpstorm/channel.py", line 326, in write_frames
    self.check_for_errors()
  File "/home/wirebot/.virtualenvs/cayzn_yield.etl/local/lib/python2.7/site-packages/amqpstorm/channel.py", line 169, in check_for_errors
    self._connection.check_for_errors()
  File "/home/wirebot/.virtualenvs/cayzn_yield.etl/local/lib/python2.7/site-packages/amqpstorm/connection.py", line 155, in check_for_errors
    raise self.exceptions[0]
AMQPConnectionError: Connection was closed by remote server: CONNECTION_FORCED - broker forced connection closure with reason 'shutdown'

On the rabbitmq server, we have the following logs: client unexpectedly closed TCP connection

Finally here is the code i use to create an amqp socket (this socket is then used by logging to send log in the queue) :

class AMQPStormSocket(object):

    def __init__(self, host, port, username, password, virtual_host, exchange, queue, exchange_is_durable,
                 queue_is_durable, exchange_type, fallback_call):

        # create connection & channel
        self.connection = amqpstorm.Connection(host, username, password, port, virtual_host=virtual_host, timeout=1)
        self.channel = self.connection.channel()

        # create an exchange, if needed
        self.channel.exchange.declare(exchange=exchange, exchange_type=exchange_type, durable=exchange_is_durable)
        # create a queue, if needed
        self.channel.queue.declare(queue=queue, durable=queue_is_durable, passive=False, auto_delete=False)
        # bind it
        self.channel.queue.bind(queue=queue, exchange=exchange)

        # needed when publishing
        self.exchange = exchange

        self.fallback_call = fallback_call

    def sendall(self, data):
        try:
            self.channel.basic.publish(body=data, exchange=self.exchange, routing_key='')
        except Exception as e:
            self.fallback_call(e)

    def close(self):
        try:
            self.channel.close()
            self.connection.close()
        except Exception:
            pass

Do you have an idea on how to fix these errors ?

I use RabbitMQ, all the excahnges and consumers were preconfigured with differnet options. Unfortunately when I try to consume from a durable queue I get

mqpstorm.exception.AMQPChannelError: Channel 1 was closed by remote server: NOT_FOUND - no previously declared queue

If I try to declare a queue without specifying any params I get

Channel 1 was closed by remote server: PRECONDITION_FAILED - inequivalent arg 'durable' for queue 'test_queue' in vhost 'lizziebot': received 'false' but current is 'true'

The only way is to specify all the params including including ttl, autodelete, etc. I cannot use this library because I need rabbitmq structure to be configured not on the client side ;-(

Does the module supports AMQP-1-0-0.

I do not see anything regarding supported version in the Doc.

If not, is there any plan to support it in the future?

I am using Python3 version 3.10.4 on Ubuntu 18.04 LTS and encountering the following error when attempting to install amqpstorm:

$ pip install amqpstorm
Keyring is skipped due to an exception: module 'collections' has no attribute 'MutableMapping'
Defaulting to user installation because normal site-packages is not writeable
Collecting amqpstorm
  Using cached AMQPStorm-2.10.4.tar.gz (71 kB)
  Preparing metadata (setup.py) ... error
  error: subprocess-exited-with-error

  × python setup.py egg_info did not run successfully.
  │ exit code: 1
  ╰─> [20 lines of output]
      Traceback (most recent call last):
        File "<string>", line 2, in <module>
        File "<pip-setuptools-caller>", line 14, in <module>
        File "/usr/lib/python3/dist-packages/setuptools/__init__.py", line 12, in <module>
          import setuptools.version
        File "/usr/lib/python3/dist-packages/setuptools/version.py", line 1, in <module>
          import pkg_resources
        File "/usr/lib/python3/dist-packages/pkg_resources/__init__.py", line 77, in <module>
          __import__('pkg_resources.extern.packaging.requirements')
        File "/usr/lib/python3/dist-packages/pkg_resources/_vendor/packaging/requirements.py", line 9, in <module>
          from pkg_resources.extern.pyparsing import stringStart, stringEnd, originalTextFor, ParseException
        File "<frozen importlib._bootstrap>", line 1027, in _find_and_load
        File "<frozen importlib._bootstrap>", line 1006, in _find_and_load_unlocked
        File "<frozen importlib._bootstrap>", line 672, in _load_unlocked
        File "<frozen importlib._bootstrap>", line 632, in _load_backward_compatible
        File "/usr/lib/python3/dist-packages/pkg_resources/extern/__init__.py", line 43, in load_module
          __import__(extant)
        File "/usr/lib/python3/dist-packages/pkg_resources/_vendor/pyparsing.py", line 943, in <module>
          collections.MutableMapping.register(ParseResults)
      AttributeError: module 'collections' has no attribute 'MutableMapping'
      [end of output]

  note: This error originates from a subprocess, and is likely not a problem with pip.
error: metadata-generation-failed

× Encountered error while generating package metadata.
╰─> See above for output.

note: This is an issue with the package mentioned above, not pip.
hint: See above for details.

I have seen recommendations to upgrade to the latest version of requests when this happens, but I already have the latest version of requests installed, here is the contents of my requirements.txt file:

boto3==1.23.0
botocore==1.26.0
jmespath==1.0.0
python-dateutil==2.8.2
s3transfer==0.5.2
six==1.16.0
urllib3==1.26.9
redis==4.3.1
requests==2.27.1
amqpstorm==2.10.4
pyyaml==6.0

Hello,

Thanks for your work on this library 👍 I have a question regarding Rabbitmq memory watermarks and the way amqpstorm handles them.

Imagine the following:

1. I have a process that keeps publishing messages (basically an infinite loop for this example)
2. At a certain point memory alarm is triggered on RMQ, and further publishes are blocked (Writes on a blocked connection will time out or fail with an I/O write exception.), also the publisher get notified that a connexion is blocked (Compatible AMQP 0-9-1 clients will be [notified] when they are blocked and unblocked.).
3. Then Rabbitmq flushes out some of the data to disk, and unblocks publisher

What happens on AMQPstorm side ? Does it raise ? do we lose some messages ? does the block/unblock logic trigger some logic in AMQPStorm ?

Thanks for the help

Links: RMQ doc

Hi,

First, thanks for the work.

We are experiencing some slow-down while enqueuing message with confirm-delivery. Some time before we were not using this feature and enqueuing would be quite fast, and after enabling this feature, enqueuing would become quite slower.

I checked and it seems that the lib wait for the confirmation on each message while on the following blog post (https://blog.rabbitmq.com/posts/2011/02/introducing-publisher-confirms) they seems to tell us to enqueue all the message and then wait for all the confirmations, do you think there would be a way for amqpstorm to handle that ? If you have some indications, I could try an implementation.

Thank you.

Hi,

This is more of a question than an issue - is it possible to be able to manage a long-lived connection such that a reconnection can be immediately attempted if the channel/connection is closed?

I see the robust consumer example, but nothing for a connection that maybe used just for publishing. I don't see the ability to add any callbacks for a closed connection or a way to block the current thread until there's an error on the connection/channel, or maybe I'm just missing something.

Thanks for any help in advance.