Last Value queue

[pgarbo]

Hi,

I just came across your library, and it seems to be much more developer friendly than AMQPNetLite and there's good documentation so looks like a good candidate to replace bare AMQPNetLite which I just started experimenting with.
I have specific requirement to create and use "last value queue" (as described here: https://activemq.apache.org/components/artemis/documentation/latest/last-value-queues.html) from code - is it possible using this library?

thanks,
Pawel

[Havret]

Hi @pgarbo,

It's not possible at the moment, but that's sth I've got on my radar for a while. I will try to implement it over the weekend.

Update
Please check 2.11.0-preview1

It should be as simple as:

var endpoint = Endpoint.Create("localhost", 5672, "guest", "guest");
var connectionFactory = new ConnectionFactory();
var connection = await connectionFactory.CreateAsync(endpoint);
var topologyManager = await connection.CreateTopologyManagerAsync();
await topologyManager.CreateQueueAsync(new QueueConfiguration
{
    Address = "my-address",
    Name = "my-queue",
    Durable = true,
    AutoCreateAddress = true, // true if address doesn't exist

    LastValueKey = "my-last-value-key" // set last value key
});

Then, later in your producing code you have to set application property with the key that matches the pre-defined LastValueKey:

var producer = await connection.CreateProducerAsync("my-address");
await producer.SendAsync(new Message("foo1")
{
    ApplicationProperties =
    {
        ["my-last-value-key"] = "1"
    }
});
await producer.SendAsync(new Message("foo2")
{
    ApplicationProperties =
    {
        ["my-last-value-key"] = "1"
    }
});

Please let me know if you happy with this, so I can cut an official release.

Best,
Havret

[pgarbo]

Hi,

Thanks for non-destructive flag. This all works fine with anycast queues, but I am struggling to get it to work with multicast.
The general idea is that desktop clients connect to the queue, and they get one last value and while they run, they receive updates (let's say current currency rate - we dont care about historical data). Once client disconnects and connects again (eg next day), it receives one last value on start-up and waits for any updates. Same for any new client that connects for the very first time.
Is this possible? Not sure if this is issue with sender or receiver:

await topologyManager.DeclareQueueAsync(new QueueConfiguration
            {
                Address = Config.MulticastAddress,
                Name = Config.MulticastAddress,
                Durable = true,
                AutoCreateAddress = true,
                NonDestructive = true,
                RoutingType = RoutingType.Multicast,
                LastValueKey = "my-last-value-key" // set last value key
            });
var multicastProducer = await connection.CreateProducerAsync(Config.MulticastAddress);
var message = "foo"+DateTime.Now.Second;
await multicastProducer.SendAsync(new Message(message)

                {
                    ApplicationProperties =
                    {
                        ["my-last-value-key"] = "1"
                    }
                });

and the receiver:

var queueName = Config.MulticastQueue + "1";
//different options
//var consumer = await connection.CreateConsumerAsync(Config.MulticastAddress, RoutingType.Multicast); //works like regular non-durable queue, no last value on startup
//var consumer = await connection.CreateConsumerAsync(Config.MulticastAddress, queueName); //works like regular durable consumer, no last value on startup
var consumer = await connection.CreateConsumerAsync(Config.MulticastAddress, Config.MulticastAddress); //this almost works: on connect client gets the latest and gets further updates but if more than one client, each client receives updates in round robin mode
            Console.WriteLine("waiting...");
            while (true)
            {
                var message = await consumer.ReceiveAsync();
                var msg = message.GetBody<string>();
                await consumer.AcceptAsync(message);
                Console.WriteLine($"Received {msg}");
            } 

[Havret]

I'm afraid what you're trying to do (having non-destructible pub-sub with last value key) won't be possible using purely the client lib. At least not with the current version of the client. It would require some additiona work to extend TopologyManager with additional API to update address settings.

What you're looking for is described in this SO question.

You basicaly need to define retroactive address instead of non destructible queue.

To do this you need to add the following configuration to your broker.xml

<address-setting match="my-address">
   <retroactive-message-count>3</retroactive-message-count>
</address-setting>
<address-setting match="*.*.*.my-address.*.retro">
   <default-last-value-key>my-last-value-key</default-last-value-key>
</address-setting>

retroactive-message-count should correspond with the number of unique keys you can have in your flow.

Then your publisher code might look like that:

using ActiveMQ.Artemis.Client;
var endpoint = Endpoint.Create("localhost", 5672, "artemis", "artemis");
var connectionFactory = new ConnectionFactory();
var connection = await connectionFactory.CreateAsync(endpoint);

var producer = await connection.CreateProducerAsync("my-address", RoutingType.Multicast);

for (int i = 0; i < 1000; i++)
{
    var key = (i & 3) switch
    {
        1 => "a",
        2 => "b",
        _ => "c"
    };

    await producer.SendAsync(new Message($"id='{i}' key = {key}")
    {
        ApplicationProperties =
        {
            ["my-last-value-key"] = key
        }
    });
}

And the consumer:

using ActiveMQ.Artemis.Client;
var endpoint = Endpoint.Create("localhost", 5672, "artemis", "artemis");
var connectionFactory = new ConnectionFactory();
var connection = await connectionFactory.CreateAsync(endpoint);

var consumer = await connection.CreateConsumerAsync(address: "my-address", RoutingType.Multicast);

while (true)
{
    var msg = await consumer.ReceiveAsync();
    Console.WriteLine(msg.GetBody<string>());
}

When the new consumer subscribes to the address, it will get the snaphot of last values for all unique keys and all subsequent updates.

Best,
Havret

[pgarbo]

Hi,

Thanks for the details - I really appreciate your effort. I will try retroactive approach; however, I have found a workaround for now. Please let me know if you see any issue with this approach:

1. Create anycast durable non-destructive last-value queue for initial values
2. Create multicast non-durable queue for updates.

The scenario is as follows:

• Server always publishes to queue 1 and 2.
• Client connects to anycast queue 1 on start-up to get the latest value and then listens to updates on queue 2 until exit.

I have tested it with multiple clients, and it seems to work.
What do you think? Do you see any issues with that?

btw I am happy to accept your answer and confirm that anycast last-value non-destructive works fine. When do you expect to release non-preview package?

Pawel

[Havret]

Hi,

This workaround should work, however it may be tricky to discern when you should stop listening to queue 1 and switch to 2 for updates. If you keep listening to both queues all the time, you need to be careful regarding possible race conditions.

As to the release, we can either wait for #400 or I could cut the release without it.

Havret

[pgarbo]

we can wait for #400

Pawel