Building Notification Engine Using Integration Principles

While working for a transportation company, I needed to get notifications sent back and forth between many web applications and I, as would anybody expect, started searching the web for a good tool or any article explaining how to do it. I was surprised to find nothing, considering my need was very simple and common. A couple of blogs though showed some guidance and tools but nothing really complete and running. So I decided to build it. Unfortunately, I cannot share with you the product I come up with because it belongs to my client as specified in my contract. But I will try to get you ready with some steps I walked through...

By notification engine, as I laid it out in the title, I mean a piece of software that allows sending data between applications. This is integration, actually. So what I am heading to do is architecting integration. So the scope of initial need is becoming bigger and more useful for several other things I was doing "spaghetti style". So, to do integration you need to know who will send what to whom and who will listen to whom for what. This is done by having a sketch of your applications’ data flow. This sketch helped me to point out the spaghetti communication between many of our enterprise applications.

Having the big picture in mind, I decided to focus and start by implementing the sending of notifications. Here's where the message broker comes in. Message broker, a tool that most of the time is referred to as message bus (even if there is a difference), is a technology that let applications send messages and others to listen to incoming messages. Many products help achieving that and it's worth the effort to look at them. The ones I selected at first are MSMQ, ActiveMQ, ZeroMQ, RabbitMQ. I had many criterias that helped me select RabbitMQ, which is I think a terrific open source and free product.

Once the step of choosing the message broker tool is done, came another question which led to another decision: how to connect my applications to RabbitMQ. Of course RabbitMQ has a .net client but using it meant the same piece of code had to be written many times for different purposes. So I looked to the net again and selected a couple of good tools: MassTransit, EasynetQ, etc. I had many criterias in mind but the one that stood up for simplicity and documentation was EasynetQ. Masstransit is also a good product for .net world. I started using EasynetQ to implement a publish subscribe design pattern for notification purposes but was quickly disappointed the way this pattern was implemented. So I got back to .net client offered by RabbitMQ and I didn't regret it at all. Because this terrific and simple .net client gave me control and more freedom during the implementation process. Yes, I had to do some abstractions to avoid redundancy but once this is done properly, you end up having a fine-grained control over everything happening behind the scene.

Once the right tooling selection was settled, I installed them on my machine and left them aside. I started writing down and sketching my use cases. This important step helps figuring out the right configurations and approaches to take using RabbitMQ. It clarifies and defines the modes in which to communicate with queues: topic, fanout, direct, headers, etc. Should I use virtual host or not? In which case would I need persistence and expiration? What about failures and dead letters? What about errors, retrying and requeuing? Many questions rose up during the process. I took the time necessary to think about my requirements and how to deal with all those questions.

Now I am ready to put up things and build my notification engine. Let me set the technical context. I need to send a message from an MVC 5 application through a web API 2 to 2 others MVC 5 applications. So basically I have an API endpoint that will push a message (via a post http action) to the broker and the broker will hand it back to subscribers. Besides notifications, this architecture could work also for many other data flowing between applications. Sometimes, other modes should be explored depending on use case and requirements. For my notification I chose Topic mode.

So to get the notification flow done, I need to configure in RabbitMQ server

1. two queues, one for each application,
2. create a topic mode exchange
3. create the bindings to the queues. The binding is based on target applications name. So to send a message I used this simplified version of code:

   public void Produce(IProducerMessage message)
        {
           SetModul();
           var properties = GetProperties();
 
            //Serialize
           var pm = Serialize(message);
           byte[] messageBuffer = Encoding.Default.GetBytes(pm);
           
           //Send message
           _model.BasicPublish(this._mySetting.ExchangeName, this._mySetting.RoutingKey, properties, messageBuffer);
        }
 
  private string Serialize(IProducerMessage message)
        {
            var settings = new JsonSerializerSettings {ContractResolver = new CamelCasePropertyNamesContractResolver()};
            return JsonConvert.SerializeObject(message, Formatting.Indented, settings);
        }
 
  private IBasicProperties GetProperties()
        {
            //Setup properties
            var properties = _model.CreateBasicProperties();
            properties.DeliveryMode = 2;//2 persistent, 1 non-persistent
            properties.ContentType = _mySetting.ContentType;
            properties.ContentEncoding = "base 64 utf8";
            properties.CorrelationId = "m2a";
            properties.MessageId = Guid.NewGuid().ToString();
            properties.Timestamp = new AmqpTimestamp(GetUnixTime());
            //properties.ReplyTo
            properties.Expiration =_mySetting.Expiration;//ms
         
            return properties;
        }
 
  private void SetModul()
        {
            var myconnector = new myConnector(_mySetting);
            var conn = myconnector.GetConnection();
            _model = conn.CreateModel();
        }
 
 
 public class myConnector
    {
       
        private ConnectionFactory _connectionFactory;
        private IConnection _connection;
        private readonly mySettingProducer _mySetting;
 
        public myConnector(mySettingProducer mySetting)
        {
            _mySetting = mySetting;
        }
 
        public IConnection GetConnection()
        {
            SetupConnection();
            return _connection;
        }
 
      private void SetupConnection()
        {
           // ValidateSettings();
            _connectionFactory = new ConnectionFactory
            {
                HostName = this._mySetting.Host,
                UserName = this._mySetting.Username,
                Password = this._mySetting.Password
            };
 
            if (!string.IsNullOrEmpty(this._mySetting.VirtualHost)) _connectionFactory.VirtualHost = this._mySetting.VirtualHost;
            if (this._mySetting.Port > default(int)) _connectionFactory.Port = this._mySetting.Port;
 
            _connection = _connectionFactory.CreateConnection();
          
        }     
}

To subscribe an application to a specific queue I used the following simplified version of code:

public class myConsumer : IDisposable
    {
 
        private string _hostName = "my ip address";
        private string _userName = "username";
        private string _password = "password";
        private string _queueName = "queuename";
 
        private const string VirtualHost = "virtualhostname";//if any
        private int Port = 0;//specify the port if any
 
        public delegate void OnReceiveMessage(string message);
 
        public bool Enabled { get; set; }
 
        private ConnectionFactory _connectionFactory;
        private readonly IConnection _connection;
        private readonly IModel _model;
        private Subscription _subscription;
        private readonly myNotificationsender _myNotificationHub;//for signalR
 
        public myConsumer()
        {
                     this._connectionFactory = new ConnectionFactory
                                     {
                                         HostName = this._hostName,
                                         UserName = this._userName,
                                         Password = this._password
                                     };
 
             this._myNotificationHub = new myNotificationsender();//for signalR
 
            if (!string.IsNullOrEmpty(VirtualHost))
                this._connectionFactory.VirtualHost = VirtualHost;
            if (this.Port > 0)
                this._connectionFactory.Port = this.Port;
 
            this._connection = this._connectionFactory.CreateConnection();
            this._model = this._connection.CreateModel();
            this._model.BasicQos(0, 1, false);
 
        }
 
       
       public void Start()
        {
            this._subscription = new Subscription(this._model, this._queueName, false);
 
            var consumer = new ConsumeDelegate(this.Pull);
            consumer.Invoke();
        }
        private delegate void ConsumeDelegate();
 
 
        private void Pull()
        {
 
            var consumer = new EventingBasicConsumer(this._model);
            consumer.Received += (o, e) =>
                {
                    var data = Encoding.UTF8.GetString(e.Body);
                    this._myNotificationHub.PushNotificationsToClients("", data);//for signalR
                  
                };
 
            var consumerTag = this._model.BasicConsume(this._queueName, true, consumer);// true is aknowledge that the message is received
      
        }
   }

The last thing left is a way to get the notification displayed to users without any page refresh. For this I need Signalr. Signalr will notify the view using JavaScript for any coming message. See the commented out code for signalR. You still need the following server side:

public class myNotificationsender
    {
        public void PushNotificationsToClients(string name, string data)
        {
            try
            {
                var context = GlobalHost.ConnectionManager.GetHubContext<myNotificationHub>();
                context.Clients.All.addNewMessageToPage(data);
            }
            catch (System.Exception)
            {
                // do something with the exception as logging
                return;
            }
        }
    }

And finally in layout view or any view use the following javascript to wire up the signalR code:

var notHub = $.connection.myNotificationHub;
notHub.client.addNewMessageToPage = function(message) {
    // Add the message to the page.
    $('#notifbell').css("display", "");
    $('#notif').html(message);
};
$.connection.hub.start();

Of course, this is just an introduction because you still need to configure the Deadletters, routing failures and the retry policy if you need any. The serialization and large messages should also be taken in consideration. If you have to expose your broker to the outside world, you should also take care of the TLS to secure the data exchange.

I hope this introduction will give you a sense and guidance for starting integration implementation.