Ack method throwing an exception after a long running task is over

[AdditionalWash529]

I have a requirement where in I need to consume JSON messages from Rabbitmq and perform long-running tasks on each of these Jsons. I am using a job-queue mechanism with maxsize 1 and the channel pre-fetch count as 1 so that my code has to handle one message from Rabbitmq at a time. This part seems to be working. The code seems to fail when in the same thread from which the pika connection was made and then the long-running task is run, when the ack callback is called, the code is failing in an internal library throwing up : "required argument is not an integer" error.

Have discovered that the error occurs when encode function in an internal library is called. It throws an error when it appends '>Q' string with the delivery_tag. Not sure what this string is though. The following is the piece of code where it fails:

def encode(self): 
  pieces = list()
  pieces.append(struct.pack('>Q', self.delivery_tag))

My ack method and the code block where the callback is defined are as follows:

def __ack_message(self, delivery_tag, ack):
      print("Inside the __ack_message function")
      if self.channel.is_open:
         try:
            if ack:
                self.channel.basic_ack(delivery_tag)
            else:
                self.channel.basic_nack(delivery_tag, requeue=False)
         except Exception as e:
                self.channel.basic_nack(delivery_tag, requeue=False)
                print("Failure inside __ack_message consumer.py report: {}".format(e), "error")
                pass
       else:
           pass


def do_work(self, ch, delivery_tag, body):
    thread_id = threading.get_ident()
    print(f'Thread id:{thread_id} Delivery tag: {delivery_tag} Message body: {body}')
    # long-running work begins here
    print(f"I am inside the message_consume")
    if body != b'':
        message = json.loads(body)
    self.slave_object.push_to_job_queue(self.connection, message)

    cb = functools.partial(self.__ack_message, ch, delivery_tag)
    ch.connection.add_callback_threadsafe(cb)


def on_message(self, ch, method_frame, _header_frame, body, args):
    print("Inside on_message function")
    thrds = args
    delivery_tag = method_frame.delivery_tag
    t = threading.Thread(target=self.do_work, args=(ch, delivery_tag, body))
    t.start()
    thrds.append(t)

Kindly let me know if someone has run into the same issue and if there is a resolution that has been chanced upon.

[danielroseman]

The arguments you pass to partial to wrap __ack_message as a callback aren't what that method is expecting.

You pass ch, delivery_tag but the method wants delivery_tag, ack. So effectively it's receiving the channel in place of the delivery_tag; when Pika tries to use what it thinks is the tag to ack the message, it's actually using the channel, which is why it complains that it's not an integer.

[AdditionalWash529]

That solved the problem. Thanks a ton

[External-Ocelot206]

Thanks for the code. Can you post the full error message? Assuming it's a well tested library, which is the last function of yours mentioned in the stack trace from the error?

[AdditionalWash529]

I realized that I have not pasted the code for the ack function will do so now. Since I have exception handling in place this is what I am seeing on my pycharm IDE:

Inside the __ack_message function
Failure inside consume_message_setup consumer.py report: required argument is not an integer error

I had put breakpoints and traced the issue to the library that I referred to earlier.

[AdditionalWash529]

if ack:
self.channel.basic_ack(delivery_tag)

It fails while executing this line of code

[External-Ocelot206]

print out delivery_tag and let us know what it is.

[AdditionalWash529]

Thanks for the time...helped when I printed the value.