0
\$\begingroup\$

Use case :
I have a Order web Api which creates an orders and publish the order to messaging queue. Notification service (web api 2) consumes this message by subscribing to the messaging queue and notify the user.

I'm newbie in messaging queue
I'm using cloudAMQP as queue service.

Order Service where i publish the event

public class MessageQueueConfig
{
    public static IModel Channel { get; private set; }
    public static IConnection Connection { get; private set; }
    const string ExchangeName = "Notifications";
    public const string CreatedRoutingKey = "WorkOrderCreated";
    public const string UpdatedRoutingKey = "WorkOrderUpdated";

    public static void Register()
    {
        var factory = new ConnectionFactory()
        {
            Uri = new Uri(@"amqp://myurl"),
        };
        factory.RequestedHeartbeat = 5;
        Connection = factory.CreateConnection();
        Channel = Connection.CreateModel();
        Channel.ExchangeDeclare(exchange: ExchangeName, type: ExchangeType.Topic, durable: true);
    }

    public static void Publish(string routingKey, object body)
    {
        var byteBody = MemorySerializer.ObjectToByteArray(body);
        if (Connection.IsOpen)
        {    
            Channel.BasicPublish(exchange: ExchangeName,
                    routingKey: routingKey, basicProperties: null,
                    body: byteBody);
        }
    }
}

Order Service Global.asax.cs

protected void Application_Start()
{
  FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
  //web api config
  MessageQueueConfig.Register(); //where i start the queue.
}

Order Service Controller

[HttpPost]
public async Task<IHttpActionResult> InsertOrder(Order Parms)
{
   try
   {

    var jobNumber = await _WorkOrderB.InsertOrder(Parms, passCode);
    ResponseObj responseobj = new ResponseObj();               
    var msg= new MyQueueClass{ //some data};
    MessageQueueConfig.Publish(MessageQueueConfig.CreatedRoutingKey, msg);
    responseobj.Result = GeneralDTO.SuccessStatus;
    responseobj.ResponseData = jobNumber;
    return Ok(responseobj);
   }
   catch (Exception exception)
   {
     return InternalServerError(exception);
   }
}

Notification Service where i consume the events

public class MessageQueueConfig
{
    public static IModel Channel { get; private set; }
    public static IConnection Connection { get; private set; }
    const string ExchangeName = "Notifications";
    const string CreatedRoutingKey = "WorkOrderCreated";
    const string UpdatedRoutingKey = "WorkOrderUpdated";
    static List<string> RoutingKeys = new List<string>
    {
        "WorkOrderCreated",
        "WorkOrderUpdated"
    };

    public static void RegisterAndSubscribe()
    {
       var factory = new ConnectionFactory()
       {
            Uri = new Uri(@"amqp://myurl"),
       };
       factory.RequestedHeartbeat = 5;
       Connection = factory.CreateConnection();
       Channel = Connection.CreateModel();
       Channel.ExchangeDeclare(exchange: ExchangeName, type: ExchangeType.Topic, durable: true);
       var queueName = Channel.QueueDeclare().QueueName;

       foreach (var routingKey in RoutingKeys)
        {
            Channel.QueueBind(queue: queueName,
                          exchange: ExchangeName,
                          routingKey: routingKey);
        }

        var consumer = new EventingBasicConsumer(Channel);
        consumer.Received += Consumer_Received;
        Channel.BasicConsume(queue: queueName,
                             autoAck: true,
                             consumer: consumer);
  }
 async static void Consumer_Received(object sender, BasicDeliverEventArgs e)
    {
        switch (routingKey)
        {
            case CreatedRoutingKey:                   
                await ProcessWorkOrderCreated(e.Body);
                break;
            case UpdatedRoutingKey:
                await ProcessWorkOrderUpdated(e.Body);
                break;
            default:
                break;
        }

Notifcation Service Global.ascx.cx

protected void Application_Start()
{
    //Web api Conifg
    BundleConfig.RegisterBundles(BundleTable.Bundles);
    MessageQueueConfig.RegisterAndSubscribe();
}

Questions

  1. I would like to know any potential issue for this architecture ?
  2. What should be ideal factory.RequestedHeartbeat = 5; ?
  3. What would happen if my notification service (consumer) goes down ? i have put my queue as durable= true assuming it would still available when my consumer is back online.
  4. I haven't disposed them, Is Global.asxs.cs public void Dispose() is the right place to Close the rabbitmq connection ?

Global.asxs.cs

public override void Dispose()
  {
      MessageQueueConfig.Dispose();
      base.Dispose();
  }

Note: This code is running dev environment, i'm planning to release this next week to production

\$\endgroup\$

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Browse other questions tagged or ask your own question.