Schedule Message Delivery using AWSMQ ActiveMQ and MassTransit- 3

In this article, I am going to talk about a simple publisher/subscriber scenario for schedule message delivery using Amazon MQ, which is an implementation of Apache ActiveMQ message broker service. To impalement publisher and subscriber I used MassTransit framework.

Prerequisites & Setup

Basic knowledge on AWS Cloud service and basic understanding ASP.NET Core WebAPI and .NET Core .

Basic understanding of the event-driven communication between microservices read my pervious article (Link) and MassTransit Framework (Link)

Also read my pervious article for more understand Implement an event-driven microservice architecture on Kubernetes with RabbitMQ using MassTransit Part — 2

Source Code

You can download the source code from my Git Repo.

Use Case

Assume that we have an application, using that application user can create an email notification with scheduled time then publisher will publish email message to the queue, then consumer consume that message given scheduled time and trigger the email notification to the end-user.

If the consumer consumes the message if there is a failure or exception occur system should be able to re-try and if the re-try unsuccessful then move the message to the Error queue.

Amazon MQ

Amazon MQ is a managed message broker service for Apache ActiveMQ and RabbitMQ that makes it easy to set up and operate message brokers on AWS. Amazon MQ reduces your operational responsibilities by managing the provisioning, setup, and maintenance of message brokers for you. (Link)

Apache ActiveMQ

Apache ActiveMQ (Active Message Queuing) is a message queue service. It is an open-source, multi-protocol Java-based messaging server (JMS).

ActiveMQ provide flexibility to send messages through both queues and topics using a single Broker. In point-to-point messaging, the Broker acts as a load balancer by routing each message from the queue to one of the available consumers in a round-robin pattern. When you use pub/sub messaging, the Broker delivers each message to every consumer that is subscribed to the topic.

ActiveMQ from version 5.4 has an optional persistent scheduler built into the ActiveMQ message broker. It is enabled by setting the broker schedulerSupport attribute to true in the Xml Configuration. An ActiveMQ client can take advantage of a delayed delivery by using the following message properties:

ActiveMQ broker setup in AWS

1.) Sign-in to AWS and navigate to the Amazon MQ console.

2.) Select broker engine page, choose ActiveMQ, and then choose next.

3.) On the Select deployment mode page, choose the Deployment mode and Storage type.

• Select Single-instance broker as Deployment Mode
• Select Amazon Elastic File System (EFS) as Storage type.

4.) Provide Configure settings

After setup development mode, next step Configure settings page

• Broker name: provide broker name
• Broker instance type: For more information
• Provide User Name and Password for access ActiveMQ cluster.

5. Additional settings configuration

In the addition setting you can configure broker network and security ,public accessibility and cloud watch . Here I will be use default settings, It can access ActiveMQ console publicly. If you want restrict the access, provide the VPC ,subnet , encryption and security group. After adding necessary configuration click Create broker button to create broker.

It will take around 15mins to create activemq broker. Once broker create click on the broker and navigate to details page.

6.) ActiveMQ Web Console

In the broker details page you can see ActiveMq Web Console link ,click on the link ,It will redirect to the ActiveMQ Web console. Click on manage console link, enter your username and Password that we create step 4.

7.) Enable Scheduler plugin.

If you navigate to the scheduled section in ActiveMq console, you can see “scheduler not started!” message. To enable scheduler support edit the xml configuration add schedulerSupport attribute to true .

On the Edit Broker page, in the Configuration section, select a Configuration and a Revision and then choose Edit. Update configuration by adding schedulerSupport=”true”

Save the changes by adding description.

Navigate to the ActiveMQ broker page and click on the Edit button. On the Configuration section select the Revision version that we create.

Click on the Schedule modifications button ,click on the Immediately option it will reboot the broker right away.

After reboot complete, you can navigate to the ActiveMQ console then scheduled section you can see following table that mean your ActiveMq broker now able to handle schedule message .

Publisher Application

1.) Create Asp.net core web Api application using visual studio.

In here I am not going to describe in details how to create .net core application. You can download the source code in my GitHub repository as I mention above.

After create application install the following Nuget Packages.

2.) Add ActiveMQ connection string to appsettings.json.

provide ActiveMq host name xxxxxxx.mq.<reagion>.amazonaws.com, username and password that we create previous section.

3.) Register the ActiveMQ and MassTransit dependencies.

Here I register all dependencies as extensions inside the ConfigureServices method in the Startup.cs. Use the built-in transport message delay to schedule messages.

4.) publish schedule message to the Queue.

Create ProducerService class and inherit IProducerService interface.

A publish endpoint lets the underlying transport determine the actual endpoint to which the message is sent. In here we are using IMessageScheduler.ScheduleSend method to schedule message in the ActiveMQ.

• DestinationAddress : The destination address where the schedule message should be sent.
• ScheduledTime : The time at which the message should be delivered to the queue
• Message : The message that need to publish.

5.) Create Controller class Post method to expose the publisher service to the outside.

Consumer Application

1. ) Create Asp.net core web Api application and add Nuget packages using visual studio.

Delete Controller classes and UseRouting , UseAuthorization and UseEndpoints Configure method in Startup.cs class.

After create application add following Nuget Packages.

2.) Add ActiveMq connection string and Email credentials to appsettings.json.

3.) Register the ActiveMQ and MassTransit dependencies.

4.) ActiveMQ Consumer Message Filter and Re-try

MassTransit message filter , exception management and re-try you can read my previous article (Link) or check source code and MassTransit official documentation.

5.) ActiveMQ Consumer class.

** Download the source code and customize according to your scope. Github repository mention above.

Execute Publisher and Consumer

Once you run the consumer application, MassTransit will create queue and topic for you. If there any error or exception occur MassTransit will create error queue to manage error messages as well.

Execute the publisher application , Here I attached the sample post-man collection to execute the publisher controller. I put future time for scheduled time and rest of the details sample values.

After execute the publisher controller, It will send the message to the ActiveMq, you can see the schedule messages in ActiveMq console in scheduled section. Once it satisfied the next scheduled time, consumer will consume the message and send the email to the relevant party.

Conclusion

In this article, We setup AWSMQ ActiveMQ broker and enable schedule message option and also implement simple .Net core publisher/subscriber application to schedule message using MassTransit.

Enjoy!!! stay safe