How can I use transaction using this library

[jollyamit]

How can I use transaction using this library?
My use case is as below.

1. Create message
2. Send message to Artemis queue
3. do something else. (This step result processing failed or success)
4. Based upon outcome of step 3, I need to either commit or rollback the message sent in step 2.

[Havret]

@jollyamit The simplest example may look as follows:

using System;
using System.Threading.Tasks;
using ActiveMQ.Artemis.Client;
using ActiveMQ.Artemis.Client.Transactions;

namespace Example
{
    class Program
    {
        static async Task Main()
        {
            var endpoint = Endpoint.Create("localhost", 5672, "guest", "guest");
            var connectionFactory = new ConnectionFactory();
            await using var connection = await connectionFactory.CreateAsync(endpoint);
            await using var producer = await connection.CreateProducerAsync("my-address");
            
            await using var transaction = new Transaction();
            var message = new Message("my-payload");
            await producer.SendAsync(message, transaction);

            var result = DoSthThatMayFail();
            if (result)
                await transaction.CommitAsync();
            else
                await transaction.RollbackAsync();

        }

        private static bool DoSthThatMayFail()
        {
            var random = new Random();
            return random.Next() % 2 == 0;
        }
    }
}

You can find more examples in TransactionsSpec.cs.

[jollyamit]

@Havret, My use case is, both consumer1 and consumer2 needs to get all the messages published on a given address/topic.

Let's say if publisher publishes 1000000 messages in a day, I want all 1000000 messages to be consumed by both consumer1 and consumer2 regardless if they are online or offline. Idea is they both should get 1000000 messages if they are online or when ever they come online. (Under normal circumstance, they both should be online, but very rare case there might be possibility that one of the them or both are offline for a while)

Also I need an ability to have durable consumer subscription created dynamically/programmatically.

Based upon your comment above, if producer is up and keep pushing the messages on Topic, and let's say one of the durable consumers with subscription id (subscription-1) goes down for lets say 1 hour, and other durable consumers is still UP, why would it create the problem for broker? All the messages will be written do the disk on broker side, broker will not hold everything for durable consumers 1 with subscription-1 in the memory which is currently down... correct? Once consumers 1 with subscription-1 come back online again, broker will deliver all the messages to it.

In your example below, I tested and noticed below

1. Messages remains in the my-queue even if both consumer1 (connecting to my-address) & consumer1 (connecting to my-address) consumes the messages and commit the transaction. ( I think this is expected behavior? as consumer are not directly connecting to my-queue either using my-address::my-queue or using my-queue as subscription id ?)
2. Broker will not deliver the messages it received during the time consumer were down but it came back up lets say after 20 mins. (Messages received by broker during those 20 mins are not delivered back to consumer when it came back online.)

• my address (Multicast)
  • my-queue (Multicast)
    • consumer1
    • consumer2

[Havret]

Let's say if publisher publishes 1000000 messages in a day, I want all 1000000 messages to be consumed by both consumer1 and consumer2 regardless if they are online or offline. Idea is they both should get 1000000 messages if they are online or when ever they come online. (Under normal circumstance, they both should be online, but very rare case there might be possibility that one of the them or both are offline for a while)

In this case I would create the following topology:

- my address (Multicast)
  - my-queue1 (Multicast)
     - consumer(1) (service1)
     ...
     - consumer(n) (service1)
  - my-queue2 (Multicast)
     - consumer(1) (service2)
     ...
     - consumer(n) (service2)

A single multicast address, and two queues under this address for two types of your consumers (service1, service2). In this configuration you have two different queues under the same address, so the broker will deliver all messages to my-queue1 and my-queue2. You can attach as many consumers as you like to these queues (scale out) to ensure high availability. If you have a requirement that only a single consumer of each type should be able to process messages, you can mark the queue as Exclusive. The broker will pick a single consumer to get all the messages from a queue. If that consumer fails, the broker will auto failover and choose another consumer. For most of my use cases I just run many parallel consumers with message grouping enabled.

An example implementation may look like this --> https://github.com/Havret/activemq-artemis-net-example

Based upon your comment above, if producer is up and keep pushing the messages on Topic, and let's say one of the durable consumers with subscription id (subscription-1) goes down for lets say 1 hour, and other durable consumers is still UP, why would it create the problem for broker? All the messages will be written do the disk on broker side, broker will not hold everything for durable consumers 1 with subscription-1 in the memory which is currently down... correct? Once consumers 1 with subscription-1 come back online again, broker will deliver all the messages to it.

This is not that simple. Please check this out --> https://activemq.apache.org/components/artemis/documentation/latest/paging.html

[jollyamit]

Thanks @Havret for detailed explanation. Basically in your above example my-queue1 & my-queue2 are nothing but a dynamically created queues by the below mentioned DurableSubscriptions example program. In guess with your Topology manager class same behavior can be achieve. As you explained in https://github.com/Havret/activemq-artemis-net-reliable-pub-sub example with AutoDelete = false in the QueueConfiguration.

https://github.com/apache/activemq-artemis/blob/master/examples/protocols/amqp/dotnet/DurableSubscriptions/Program.cs

[Havret]

Basically yes, the difference is that there is no static connectionId/containerId which may cause problems during connection recovery, and of course you have full control over how the queue is configured.

[jollyamit]

Correct.. That's why I was thinking to add a support in your lib to be able to provide connectionId/containerId instead of GUID :)