Last value cache w/ dotnet API

[Matt Wilson <ma...@matlockcapital.com>]

Hello,
I'm using the dotnet API, client 0-10 0.5 release against a compatible C++ broker running on RedHat. My goal is dissemination prices. The behavior I'm looking for is for the publisher to send messages to a well known queue and for subscribers to get the last price sent followed by any subsequent updates to the price sent.

One solution that gets the behavior I want is the following:

// Publisher, on a per instrument basis (!)
var arguments = new Dictionary<string,object>() { { "qpid.ive", 1 } };
session.exchangeDeclare( instrumentSpecificExchange, "topic", null, arguments);

// For each message
message.clearData();
message.appendData( buffer );
session.messageTransfer(instrumentSpecificExchange, instrumentSpecificExchange, message );

// Subscriber
var queue = instrumentSpecificExchange;
var arguments = new Dictionary<string,object>() { { "qpid.max_count", 1 }, { "qpid.policy_type", "ring" } };
session.queueDeclare( queue, null, arguments );
session.exchangeBind( queue, queue, queue );

session.attachMessageListener( receiver, queue );

session.messageSubscribe( queue, queue, MessageAcceptMode.EXPLICIT, MessageAcquireMode.NOT_ACQUIRED, null, 0, null );
session.messageSetFlowMode( queue, MessageFlowMode.WINDOW );
session.messageFlow( queue, MessageCreditUnit.BYTE, ClientSession.MESSAGE_FLOW_MAX_BYTES );
session.messageFlow( queue, MessageCreditUnit.MESSAGE, Int32.MaxValue);

My issue with this approach is that it creates an enormous number of exchanges and just feels wrong given the last value queue feature. Before arriving at the approach above, I tried out LVQ but was unable to get the behavior I wanted. The setup:

// Publisher
var exchange = "amq.direct";
var routingKey = "prices_routing_key";
var queue = "prices.ibm.options";

var arguments = new Dictionary<string,object>() { { "qpid.last_value_queue", 1 } };
session.queueDeclare( queue, null, arguments );
session.exchangeBind( queue, exchange, routingKey );

// For each message
message.clearData();
message.appendData( buffer );
message.ApplicationHeaders["qpid.LVQ_key"] = instrumentSpecificKey;
session.messageTransfer(exchange, message );

// Subscriber
var queue = "prices.ibm.options";

session.attachMessageListener( receiver, queue );

session.messageSubscribe( queue, queue, MessageAcceptMode.EXPLICIT, MessageAcquireMode.NOT_ACQUIRED, null, 0, null );
session.messageSetFlowMode( queue, MessageFlowMode.WINDOW );
session.messageFlow( queue, MessageCreditUnit.BYTE, ClientSession.MESSAGE_FLOW_MAX_BYTES );
session.messageFlow( queue, MessageCreditUnit.MESSAGE, Int32.MaxValue);

Essentially, when I set a queue to be an LVQ the very first client gets the last values and then receives updates as they come in. The second client gets ALL the messages the first client received and then subsequent updates. This is a problem when there would be several hundred thousand updates sent in a given day, both on the broker and client. Changing to LVQ_NO_BROWSE (qpid.last_value_queue_no_browse) in the above example, the first client gets the last values but then does not receive any further updates. The second client to the same queue receives whatever updates happened between when the first client came up and when it came up as the last values (and also gets no further updates).

My questions:
1) What's a best practice/approach for doing a price feed with AMQP? Should I be using last value queues? Does it make sense to publish prices to 2 queues, a last value queue (LVQ_NO_BROWSE) and an "event" queue (LVQ with TTL set to some short value)?
2) With last value queues, is there a way to get only the last value and then updates when additional clients connect?
3) With last value queues no browse, is there a way to continue to receive updates after the first batch of values?

Any help or direction would be greatly appreciated!

Thanks,
Matt Wilson
Matlock Capital Management
matt@matlockcapital.com<ma...@matcap.com>