Listening to messages with given routing key

[alanxz]

Listening to messages with given routing key

[alanxz]

Unfortunately those examples are all out of date (they are relics from v1 of the SimpleAmqpClient library). See issue #3.

An example that would do what you want would look like:

#include <SimpleAmqpClient/SimpleAmqpClient.h>

using namespace AmqpClient;

int main()
{
  // You define this, also assuming its been declared on the broker
  const std::string exchange; 
  const std::string routing_key; // You define this also

  // Probably want to use the correct overload to connect to your own broker
  Channel::ptr_t channel = Channel::Create(); 

  // Declare a queue, let the broker take care of giving the queue a name
  const std::string queue = channel->DeclareQueue("");

  channel->BindQueue(queue, exchange, routing_key);

  const std::string consumer = channel->BasicConsume(queue, "");

  // This will wait until there's a message in the queue
  Envelope::ptr_t envelope = channel->BasicConsumeMessage(consumer);


  return 0;
}

I'd be interested in a (short) repro of the example that causes a bad_alloc exception to be thrown.

[adambadura]

I used same code (well, similar enough).

If the BindQueue call is done exception it thrown when message is routed. If I comment it out BasicConsumeMessage never exits (even after routing a message).

I will try to provide more details but I'm afraid it would be possible only tomorrow or even later when I'm back at work and with some extra time, since its to complicated to get it working here (RabbitMQ, MinGW, CMake, ...).

[alanxz]

Channel::Create(), DeclareQueue, BindQueue, and BasicConsume all do a round trip to the broker, and block while doing so, however if the broker and network is operating correctly these usually don't end up blocking for that long. Unless something is wrong with your broker or the network between your client or broker none of those should block. If they do I'd be interested to see a stack trace of where they block.

The particular overload of BasicConsumeMessage I selected above blocks indefinitely for a message from the broker. There is another overload that takes a timeout parameter that'll return a boolean indicating whether or not

Anyhow I think what would help me help you would be for you to post a succinct example of your code here.

[adambadura]

The very same configuration (machine and server) does work with examples/amqp_listen.c in alanxz/rabbitmq-c.

Tomorrow I will recheck both to cases and provide the source code I'm using. I could easily provide as well a stack trace but SampleAmqpClient is not understandable for debugger as it is (apparently) missing required debug info. I will try to build the code along the library in single solution so that I will get more useful stack trace into the library itself but that I cannot promise.

[adambadura]

More details come. My code looks like this:

void Listener::run()
{
  std::ofstream file( "C:\\Users\\a.badura\\Desktop\\Log.txt" );
  file << "Listener" << std::endl;
  file << "\t" << "host: " << theConnectionData.host << std::endl;
  file << "\t" << "port: " << theConnectionData.port << std::endl;
  file << "\t" << "user: " << theConnectionData.user << std::endl;
  file << "\t" << "password: " << theConnectionData.password << std::endl;
  file << "\t" << "vHost: " << theConnectionData.vHost << std::endl;
  file << "\t" << "routingKey: " << theConnectionData.routingKey << std::endl;
  file.flush();

  try
  {
    file << "AmqpClient::Channel::Create" << std::endl;
    file.flush();
    AmqpClient::Channel::ptr_t channelPtr =
      AmqpClient::Channel::Create(
        theConnectionData.host,
        theConnectionData.port,
        theConnectionData.user,
        theConnectionData.password,
        theConnectionData.vHost
      );

    file << "DeclareQueue" << std::endl;
    file.flush();
    std::string queueName = channelPtr->DeclareQueue( std::string() );
    file << "\t" << "queue: " << queueName << std::endl;
    file.flush();

    file << "BindQueue" << std::endl;
    file.flush();
    channelPtr->BindQueue(
      queueName,
      theConnectionData.exchange,
      theConnectionData.routingKey
    );

    file << "BasicConsume" << std::endl;
    file.flush();
    std::string consumer_tag =
      channelPtr->BasicConsume( queueName, std::string() );
    file << "\t" << "tag: " << consumer_tag << std::endl;
    file.flush();

    file << "BasicConsumeMessage" << std::endl;
    file.flush();
    AmqpClient::Envelope::ptr_t envelopePtr =
      channelPtr->BasicConsumeMessage( consumer_tag );
    file << "\t" << "body: " << envelopePtr->Message()->Body() << std::endl;
    file.flush();

    file << "BasicCancel" << std::endl;
    file.flush();
    channelPtr->BasicCancel( consumer_tag );
  }
  catch ( AmqpClient::AmqpException const& exception )
  {
    file << "Exception: " << typeid( exception ).name() << std::endl;
    file << "\t" << "what: " << exception.what() << std::endl;
    file << "\t" << "is_soft_error: " << exception.is_soft_error() << std::endl;
    file << "\t" << "reply_code: " << exception.reply_code() << std::endl;
    file << "\t" << "class_id: " << exception.class_id() << std::endl;
    file << "\t" << "method_id: " << exception.method_id() << std::endl;
    file << "\t" << "reply_text: " << exception.reply_text() << std::endl;
  }
  catch ( std::exception const& exception )
  {
    file << "Exception: " << typeid( exception ).name() << std::endl;
    file << "\t" << "what: " << exception.what() << std::endl;
  }
  catch ( ... )
  {
    file << "Exception: ..." << std::endl;
  }
}

Now in the test case there is a remote machine on which RabbitMQ runs with proper settings. My code runs as a Windows service. In addition I run from console the mentioned amqp_listen example. Message publishing is done using the RabbitMQ web management application.

If I publish a message with different routing key nothing happens. Just as expected.

But if I publish a message with my routing key the amqp_listen example in console properly receives it. While the above code closes with bad_alloc exception.

The log looks like:

Listener
    host: 10.40.0.165
    port: 5672
    user: guest
    password: guest
    vHost: /
    routingKey: task.adobe.premiere.client1
AmqpClient::Channel::Create
DeclareQueue
    queue: amq.gen-wcLGIlQWSvnIYxxQ6u17Lb
BindQueue
BasicConsume
    tag: amq.ctag-QXbcTrMiR_yJ1sY3iHTFvs
BasicConsumeMessage
Exception: class std::bad_alloc
    what: bad allocation

While stack trace at the moment of exception is following (Visual Studio 2010):

KernelBase.dll!74f2c41f()   
[Frames below may be incorrect and/or missing, no symbols loaded for KernelBase.dll]    
KernelBase.dll!74f2c41f()   
SimpleAmqpClient.2.dll!0f8dd562()   
SimpleAmqpClient.2.dll!0f8dd4e0()   
SimpleAmqpClient.2.dll!0f8a55c6()   
SimpleAmqpClient.2.dll!0f88f490()   
SimpleAmqpClient.2.dll!0f8881d3()   
SimpleAmqpClient.2.dll!0f896dca()   
SimpleAmqpClient.2.dll!0f88b72e()   
SimpleAmqpClient.2.dll!0f88b4f7()   
iFindForPremiere.exe!iFindListener::implementation::Listener::run()  Line 77 + 0x2a bytes   C++
iFindForPremiere.exe!iFindForPremiere::ListenerService::main(int __formal, int __formal)  Line 42 + 0x26 bytes  C++
iFindForPremiere.exe!service::ServiceT<wchar_t>::start(unsigned long aArgC, wchar_t * * aArgV)  Line 313 + 0x17 bytes   C++
iFindForPremiere.exe!service::application::ServiceApplicationT<wchar_t>::serviceMain(unsigned long dwArgc, wchar_t * * lpszArgv)  Line 176  C++
sechost.dll!74ca75a8()  
kernel32.dll!74f833aa()     
ntdll.dll!76f49ef2()    
ntdll.dll!76f49ec5()    

Sadly no useful stack tracing information for SimpleAmqpClient. I might try getting more useful stack trace by building the SimpleAmqpClient along my own code but it would take significant time which I don't have at the moment. But if you want that data as well let me know and I will try to provide it.

At the same time if I replace my code with an (adapted) version of the code from amqp_listen it does work well as well (thou for some reason file stream is not flushed well enough after each message and you only get the log once the service closes).

[alanxz]

Thanks for the repro. I was able to reproduce the behavior your seeing with std::bad_alloc. The issue is a bug in the way SimpleAmqpClient handles copying a table with 0 entries (which the RabbitMQ management console sends a message with a 0-length header table). When I try to allocate something of 0-length, I get a NULL pointer which I incorrectly detect as a failed allocation. I will have a fix for this shortly.

FYI: to get those stack frame names, put the SimpleAmqpClient.2.pdb somewhere where the MSVC debugger can find it.

[alanxz]

Ok, I have a fix in place in the master branch for the issue you're seeing. Give it a shot and let me know if that fixes things for you.

[adambadura]

I re-run the same test and looks like its working fine now. Thanks a lot!

[swatinair123]

Hi I am running my code in win 32 debug mode. But I am writing a simple code for sending the msg. But my code crashes at channel->BindQueue(queue, exchange, routing_key);
with the exception Microsoft C++ exception: [rethrow] at memory location 0x00000000.
icrosoft C++ exception: AmqpClient::AccessRefusedException at memory location 0x008FF080. .
Can anyone pls help?