On 10/06/2013 09:59, Gordon Sim wrote:
On 06/09/2013 07:52 PM, CLIVE wrote:
I have been working on replacing a QPID messaging application that uses
the current client C++ API, with a new version based on the C++
messaging API. But I have hit a few problems that I hope someone can
provide assistance with. Attached is a simple client application that I
have been using to demonstrate the problems I have been having (using
CentOS 6.3 64bit).
My particular use case is for a named queue with several bindings
associated with it. An easy use case that the current client API can
handle, but something the messaging API seems to struggle with.
The attached test program uses the messaging API to create two receivers
using the following address strings:
qpidRxer; {create:always, delete:always, node:{type:queue,
x-bindings:[{exchange:amq.topic, key:rxer1}]}}
qpidRxer; {create:always, delete:always, node:{type:queue,
x-bindings:[{exchange:amq.topic, key:rxer2}]}}
Using qpid-stat I can see the qpidRxer queue has been created and the
rxer1/rxer2 bindings applied. So all good so far.
So I send a test message with the spout application (from
cpp/examples/messaging) using the following options:
./spout --content 'rxer2 test message' 'amq.topic/rxer2'
This is where things start going wrong as the message is received by the
receiver instance that was used to create the rxer1 binding. If I send
the message again, then it is received by the Receiver associated with
the rxer2 binding. If I send the message several times, then the
Receivers just cycle round each taking it in turns to receive the test
message. So it would appear that the message filtering used by the
Receiver implementation does not go down to the binding key level. Which
seems inconsistent with the address strings I have used. I can program
around this, but its not ideal.
Do you actually want to have two receivers, each receiving a different
portion of messages? Or are you creating two merely to create two
bindings?
Assuming the latter, the x-bindings property is a list, so you can
specify multiple bindings in that, e.g.
qpidRxer; {create:always, delete:always, node:{type:queue,
x-bindings:[{exchange:amq.topic, key:rxer1}, {exchange:amq.topic,
key:rxer2}]}}
The application needs to dynamically create/remove bindings over time,
so I was initially creating individual receivers as I thought the
messages would get routed to the correct receiver based on the subject
key of each delivered message. Providing a list of bindings at start up
is not possible in this case.
The next problem occurs if I change the code slightly and basically
close the second Receiver instance, after it has been created, just to
simulate the need to sometimes remove a binding. Using qpid-stat I can
see that the queue has been deleted!!, even though I still have a valid
Receiver instance to this queue in my application. Yes, I know that I
have the 'delete:always' flag set in the address string, but I would
have thought that some kind of reference counting would have been
undertaken by the broker (particularly as I have two subscriptions in
the broker when both the Receivers were created.)
>
So I remove the 'delete:always' flag from both address strings and try
again (closing the second receiver after creation). Using qpid-stat I
can see that the qpidRxer queue is still there (which is good news), but
unfortunately both rxer1 and rxer2 bindings are still present on the
queue, so still not quite right.
What you want there is a binding tied to the scope of the receiver,
which you can get by specifying the x-bindings in the link section
rather than the node.
E.g.
qpidRxer; {create:always,
node:{x-declare:{auto-delete:True}},
link:{
x-bindings:[{exchange:amq.topic, key:rxer1}]
}}
qpidRxer; {create:always,
node:{x-declare:{auto-delete:True}},
link:{
x-bindings:[{exchange:amq.topic, key:rxer2}]
}}
Now when you close either receiver, the binding assoicated with that
will be removed but the queue will only be removed when both receivers
are closed.
If you set the capacity to zero on one of the receivers then it will
not prefetch any messages and you can receiver all the messages from
the queue through the same receiver.
Thanks for the info on the link binding, that would solve my initial
problem. Would the message routing also get fixed using this address
string i.e. Receiver1 created with rxer1, receives just the rxer1
messages from the queue?
But now I have got in to this state I
cannot see any way, via the messaging API, that can help me remove the
unused binding; I have to either use the client/console API via another
connection.
An unbind command can be sent via QMF over the same connection you use
for your application. That would in general be my prefered approach
for a solution that really needed to be doing dynamic manipulation of
the binding. I can send some more detail on the format of the message
needed for unbind requests if that sounds like a route you would be
keen to explore.
If you could send the detail that would be good. I can see that I will
probably re-design my implementation to use just a single receiver with
the bindings added/removed using the QMF commands. This would also
simplify my issues with the session going invalid.
I have looked through the address string documentation but couldn't seem
to find any flag/tag that would provide the desired functionality. So
for the time being I seem to be stuck with sticking with the client API.
On a related usage topic, Session's going invalid due to an error also
causes problems. If I create several Receivers/Senders using the same
Session and then try to create a Receiver for a queue that doesn't exist
in the broker, then the session reports an error. Which in its self it
not a problem, but there doesn't seem to be a distinction between fatal
errors and recoverable errors. Once a session errors, then that's it.
Any existing Receivers or Senders, created with that session, need to be
removed and recreated with a new session. It seems a bit excessive that
attempting to create a Receiver with a non existent queue would cause
such a headache.
Yes, I do sympathise. This however is the same as for the qpid::client
API which is used beneath the qpid::messaging API for 0-10. THe 0-10
protocol specifies that declaring a non-existent queue will
effectively end the session. The API didn't attempt to hide that as
recreating the session may not always be desirable in these cases
(since it can't always be completely transparent) though I do think
more could be done to make this less annoying.
The solution of creating a session for each
Receiver/Sender then seems a bit heavy weight as well.
Would anyone be able to provide any advice on how they have gotten
around these issues, or provide an Address String Hack that might solve
the problem? Any help gratefully received.
Also, are there any updates to the messaging API in the next QPID
release?
That depends what you mean. There is no real change to the API itself.
The lack of direct support for bind/unbind is in fact intentional for
example. There are some bug fixes and there is also AMQP 1.0 support
available (though as of 0.22 this is still a work in progress).
Thanks for the help Gordon, hopefully you have set me on the path to
successfully converting my application to the messaging API.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
.
---------------------------------------------------------------------
To unsubscribe, e-mail: [email protected]
For additional commands, e-mail: [email protected]
