[rabbitmq-discuss] Spying on "mandatory" messages

[Charly Hamy]

Matthew, Alvaro,

thanks for your answers and your reactivity!

I understand that the "mandatory" feature definitely lacks flexibility to be
used for one-to-many communications and that there is no easy workaround.

The "non auto-delete, non exclusive queue" solution would not be suitable to
my case, I really need messages to be dropped, and the sender to be alerted.
The Smart Proxy approach is interesting, but doesn't seem compatible with
the use of the mandatory flag, and the return alerts associated.

I've managed - with 10 lines of erlang - to let queues have an additional
boolean "spy" parameter, and have the "spy" queues deliver the mandatory
messages without this counting as a correct delivery, from the router point
of view. Twisted, but it's working. (and it's much simpler than having "spy"
bindings, the routing and delivery being two separated mechanisms).

My modifications:

*rabbit.hrl :*
*(edition of the amqqueue record, line 47)*
*
*
     -record(amqqueue, {name, durable, auto_delete, exclusive_owner = none,
spy=false,
                   arguments, pid}).


*rabbit_amqqueue.erl :*
*
*
*(line 115, spec of deliver/2 edited )*

* -spec(deliver/2 :: (pid(), rabbit_types:delivery()) -> boolean()|spy).*

*
*
*(edition of rabbit_amqqueue:declare)*

declare(QueueName, Durable, AutoDelete, Args, Owner) ->
    ok = check_declare_arguments(QueueName, Args),
    Q = start_queue_process(#amqqueue{name = QueueName,
                                      durable = Durable,
                                      auto_delete = AutoDelete,
                                      arguments = Args,
                                      spy = case lists:keyfind(<<"spy">>, 1,
Args) of
                                             false -> false;
                                             _ -> true
                                      end,
                                      exclusive_owner = Owner,
                                      pid = none}),
    case gen_server2:call(Q#amqqueue.pid, {init, false}, infinity) of
        not_found -> rabbit_misc:not_found(QueueName);
        Q1        -> Q1
    end.


*(edition of rabbit_amqqueue:deliver/2)*
*
*
     deliver(QPid, Delivery = #delivery{immediate = true}) ->
          gen_server2:call(QPid, {deliver_immediately, Delivery}, infinity);
     deliver(QPid, Delivery = #delivery{mandatory = true}) ->
          case gen_server2:call(QPid, {deliver, Delivery}, infinity) of
               spy -> spy;
               _ -> true
          end;
     deliver(QPid, Delivery) ->
          gen_server2:cast(QPid, {deliver, Delivery}),
          true.


*rabbit_amqqueue_router.erl :*
*(edition of rabbit_router:fold_deliveries/2, line 105)*

     fold_deliveries({Pid, true},{_, Handled}) -> {true, [Pid|Handled]};
     fold_deliveries({_,  spy}, {false, Handled}) -> {false, Handled};
     fold_deliveries({_,  _},{_, Handled}) -> {true, Handled}.


For the moment, I think I'll go with this :)
Thanks again for your help.

Charly
*
*

On Thu, Mar 24, 2011 at 4:16 PM, Alvaro Videla <videlalvaro at gmail.com>wrote:

> Hi,
>
> For spying messages on an RPC setup I'm working on an implementation of the
> Smart Proxy messaging pattern: http://www.eaipatterns.com/SmartProxy.htmlperhaps you could implement that too.
>
> This is what I've done so far: http://vimeo.com/20966661 which is at early
> stage of development and very alpha.
>
> Also take a look at the Wire Tap messaging pattern:
> http://www.eaipatterns.com/WireTap.html
>
> I guess most of the patterns from that book can be implemented with
> RabbitMQ, and some of them are already there but with different
> nomenclature.
>
> Hope this helps,
>
> Alvaro
>
> On Mar 24, 2011, at 4:05 PM, Matthew Sackman wrote:
>
> > Hi Charly,
> >
> > On Thu, Mar 24, 2011 at 03:35:35PM +0100, Charly Hamy wrote:
> >> For the req/res scenario, I'm using the mandatory flag on my messages,
> so
> >> that sending a request message to a disconnected component (no
> corresponding
> >> queue declared on the broker) returns an error to the sender. This works
> as
> >> expected.
> >
> > Well... you're using mandatory as a poor-man's form of resource
> > monitoring.
> >
> >> But now, I would want to plug new components to spy on those flows of
> >> requests and responses, without interfering with the ongoing
> >> communications... which is not by default compatible with my use of the
> >> mandatory flag : when a "spy" component is monitorng request/response
> >> messages, those messages end-up in its queue, then even if the actual
> target
> >> component for those messages is not present on the broker (no queue), no
> >> error is returned to the sender.
> >
> > Indeed, which is why mandatory is probably a mis-feature and should not
> > be used. Basically, AMQP does not really provide any sort of sane
> > resource monitoring. We've added in the 2.4.0 release consumer-death
> > notifications, but that's not going to help you here. You really have a
> > choice: make the queue non-exclusive and non-autodelete. Thus it always
> > stays there and will always receive messages. Then, you have your
> > clients just consume from it. The queue will do round-robining between
> > the consumers and all should be well, but you might have reasons why you
> > want the queue to be exclusive or autodelete. The other option is to use
> > some other tool to do monitoring of your consumers and to address that
> > problem outside of AMQP/RabbitMQ.
> >
> > Best wishes,
> >
> > Matthew
> > _______________________________________________
> > rabbitmq-discuss mailing list
> > rabbitmq-discuss at lists.rabbitmq.com
> > https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
>
> _______________________________________________
> rabbitmq-discuss mailing list
> rabbitmq-discuss at lists.rabbitmq.com
> https://lists.rabbitmq.com/cgi-bin/mailman/listinfo/rabbitmq-discuss
>
-------------- next part --------------
An HTML attachment was scrubbed...
URL: <http://lists.rabbitmq.com/pipermail/rabbitmq-discuss/attachments/20110325/bcfcad29/attachment.htm>