Table Of Contents
The current article might be useful for both beginners and experienced developers attempting to use NServiceBus (NSB). It contains detailed instructions for implementing distributed publisher-subscriber pattern by using ServiceMatrix. Further the article aggregates various SignalR examples available in different sources and introduces simple server-to-client message transfer. Solution architects may find this technique useful in many CQRS scenarios as a good alternative to WCF.
The CQRS (Command Query Responsibility Segregation) is a pattern which is based on using models of commands, queries, domains, see http://www.codeproject.com/Articles/555855/Introduction-to-CQRS. Typically, when it comes to specific implementation, sets of classes are created in order to establish publisher-subscriber interaction by sending messages via some type of command bus. This article suggests an architectural prototype for implementing CQRS. There are many articles describing various CQRS approaches, but all of them usually end up with a lot of code that is usually not flexible enough and sometimes overacrhitected. Further every article tries to solve different problem introducing a new paradigm, hence the lack of common approach.
In the real world publishers and subscribers usually reside on different machines requiring inter-process communication for message transfer. Let's discuss briefly some alternatives for inter-process communication as well as their pros and cons.
The WCF - provides clear API for inter-process communication with a good support in Visual Studio IDE as well as smooth integration with other Microsoft technologies. Non-HTTP bindings may introduce deployment problems (firewall issues). There is also no universal client that supports multiple browsers and platforms.
The NServiceBus, see http://particular.net, is a server-side technology based on MSMQ (asynchronous messages processing) which, unlike WCF, implements popular architectural paradigms such as Distributed Publisher-Subscriber and CQRS. NServiceBus being a server-side technique has a lack of support for client-side scripting and thus cannot be used directly from the browser.
In a real world scenario subscribers may utilize different technologies such as JavaFX, HTML, JavaScript, WPF and they may also reside on different platforms. In most cases the communication between publishers and subscribers involves firewalls making it very difficult for a server to push notifications to the client, This article addresses this issue and proposes some client-side techniques that allow pushing notifications to subscribers located behind firewalls by implementing some polling techniques.
REST HTTP vs WebSockets. REST (Representational State Transfer) is an architectural style that can be used to build software in which clients (user agents) can make requests of services (endpoints). User agents interact with resources. REST uses HTTP, an application-level protocol on top of TCP/IP, and it leverages GET, POST, PUT, and DELETE HTTP methods for building application logic.
WebSocket is a client-side technology leveraging TCP/IP. It represents a stream of bytes broken down into messages with undefined content. There is very little a framework can do without making assumptions about the content within a message. Furthermore working on the WebSocket level is arguably too low level for most applications just like most web applications today aren't programming directly to sockets. WebSocket client API is part of HTML 5, it is a wire protocol that handles the low-level handshaking, framing was just released in 2012. There are number of existing frameworks with higher level messaging APIs that use WebSocket underneath and may also rely on other fallback options (e.g. HTTP streaming, long polling, keep-alive, etc) . Some frameworks, like Socket.io and Vert.x, provide lightweight application facilities for responding to events and routing messages to specific handlers. Others, like CometD, provide channels for subscribing and receiving messages along with a built-in lightweight message broker. Another open-source framework called Laharsub provides HTTP REST API for AJAX/Silverlight/WCF based applications. Laharsub provides easy way for any application to organize internet scale message exchange using a publisher/subscriber pattern. However currently, ASP.NET SignalR is one of the most intensively developed libraries allowing pushing messages from server components to web clients in near real-time.
In this article I'm trying to introduce simple but flexible and scalable implementation of pushing messages from any cloud based server component to web clients using the following technology stack:
- NServiceBus.
- ServiceMatrix.
- SignalR
- ASP.NET MVC.
The server side is located in the cloud and interacts with a client through event handlers which are deployed in some web application. Further, received messages are distributed to different types of client applications via SignalR: client initiates communication; server uses opened socket to return data to the client; client uses keep-alive to preserve open connection. In such a way, the CQRS events processing is provided.
The given example was developed on Visual Studio 2012 Ultimate with the ASP.NET and Web Tools 2012.2 update installed. Also NServiceBus version (≥ 4.0) and ServiceMatrix latest version are mandatory.
In Package Manager Console (Tools | Library Package Manager) press Restore button to allow the NuGet to fetch missing packages. After that, rebuild the application. Further, follow the launch orders.
The ServiceMatrix is a new advanced developer tool designed for simplifying the establishment of NServiceBus interaction integrated with Visual Studio 2012. It is available on the official site of its creator, see http://particular.net or via Tools | Extentions and Updates select ServiceMatrix. One must have NServiceBus version (≥ 4.0) installed. For installation instructions follow link mentioned above. After ServiceMatrix was installed, a new project template of NServiceBus System appears in New Project dialog. See below:
Remark: One can notice that we chose by default .Net Framework 4.5 for our project, but the created project has in its properties .Net Framework 4.0. Of course, it will work with .Net Framework 4.5. Note that after every user's activity in ServiceMatrix's UI code fragments are generated and various assemblies are added. Unfortunately, is impossible yet to undo an operation so intermediate backups are recommended in order to avoid starting all over again. Despite that, in comparison to the pure manual variant, the ServiceMatrix is an amazing tool which is definitely worth learning to work with. Now, the following C# Class Libraries will be created: Contracts, Internal Messages and further, the Code. When we would add new services, commands and events, some code would be generated into these class libraries.<a name="stepA" id="A5"></a>
<a name="stepA" id="A5">Step A.</a> Adding a new end-point (or in other words, adding new NServiceBus host). Currently it can be one of 3 types: C# Console Class Library, MVC project or earlier type of Web application.
Open the PubSub__NSB_SignalR.MyPublisher\EndpointConfig.cs - file. Replace the AsA_Client with AsA_Publisher , as follows:
using System;
using NServiceBus;
namespace PubSub__NSB_SignalR.MyPublisher
{
public partial class EndpointConfig : IConfigureThisEndpoint, AsA_Publisher
{
}
}
Remark: To simplify understanding and verify what has been done, it is recommended to pay attention on the diagram as shown below:
<a name="stepB" id="A6">Step B.</a> Let's add new service named Notifications (note the updates in the solution explorer).
<img width="770" height="450" src="/KB/architecture/685278/StepB.gif" />
<a name="stepC" id="A7">Step C.</a> Add a new event named EventMessage.
<a name="stepD" id="A8">Step D.</a> Add a subscriber (the Notifications service).
<a name="stepE" id="A9">Step E.</a> Publish the event. The earlier created service and event names appear in the corresponding combo-boxes.
<a name="stepF" id="A22">Step F.</a> The diagram should appear as follows:
<a name="stepG" id="A11">Step G.</a> Add a new end point: in this example, it is the MVC NServiceBus host.
<a name="stepH" id="A12">Step H.</a> Deploy the component.
After the component was deployed, the following code will be generated. Double-click on EventMessageProcessor component as shown below:
Here the subscriber handles the sent event. This is represented in the diagram as follows:
<a name="stepJ" id="A13">Step J.</a> Now some manual code changes are required.
Double-click on the EventMessage event in ServiceMatrixDetails tab or open thought solution explorer the PubSub__NSB_SignalR.Contract\Notifications\EventMessage.cs. Add the following code to the EventMessage class:
using System;
using NServiceBus;
namespace PubSub__NSB_SignalR.Contract.Notifications
{
public class EventMessage
{
public Guid EventId { get; set; }
public DateTime? Time { get; set; }
public TimeSpan Duration { get; set; }
}
}
<a name="stepK" id="A21">Step K.</a> Add a new class named ServerEndpoint.cs into PubSub__NSB_SignalR.MyPublisher project and add the code below. The purpose of this class is to publish the event and every subscriber which has the handler for this event type will be able to receive the notification.
using System;
using NServiceBus;
using PubSub__NSB_SignalR.Contract.Notifications;
namespace PubSub__NSB_SignalR.MyPublisher
{
class ServerEndpoint : IWantToRunWhenBusStartsAndStops
{
public IBus Bus { get; set; }
public void Start()
{
Console.WriteLine("This will publish IEvent, EventMessage, and AnotherEventMessage alternately.");
Console.WriteLine("Press 'Enter' to publish a timer message.To exit, Ctrl + C");
while (Console.ReadLine() != null)
{
EventMessage eventMessage = new EventMessage();
eventMessage.EventId = Guid.NewGuid();
eventMessage.Time = DateTime.Now;
eventMessage.Duration = TimeSpan.FromSeconds(99999D);
Bus.Publish(eventMessage);
Console.WriteLine("Published event with Id {0} {1}", eventMessage.EventId, eventMessage.Time.ToString());
Console.WriteLine("==========================================================================");
}
}
public void Stop()
{
}
}
}
Remark: Currently the latest version of NServiceBus is 4.3.0. To implement it requires enter changes into all packages.config files contains into this solution; open the Package Manager Console and fetch missing packages via NuGet.
Update references on NServiceBus.dll, NServiceBus.Core.dll and NServiceBus.Host.exe. To avoid warning message because the AuditConfig section absence into the PubSub__NSB_SignalR.MyPublisher\App.config, follow to instructions on the page "Auditing With NServiceBus", see selected by red. Run the application. The Service Matrix works nice with 4.3.0 version.
As was mentioned above, as soon as message sent by NServiceBus arrives to the client side it will face with firewall. The SignalR "push" notification down to the client via persistent connection which client must initiate. In this paragraph ASP.NET SignalR will be added to the existing NServiceBus (ServiceMatrix) project. Verify that at least ASP.NET and Web Tools 2012.2 update is already installed.
Open the Package Manager Console via Tools | Library Packages Manager and fetch through NuGet the SignalR development packages:
PM> Install-Package Microsoft.AspNet.SignalR.Core -Version 1.0.0-rc2 –Pre –ProjectName PubSub__NSB_SignalR.SignalRSubscriber1
Remark: Currently, the latest version is SignalR 2.0. However it is oriented for Visual Studio 2013. This example uses the previous SignalR version 1.0.0-rc2 since it is compatible with the latest NServiceBus (ServiceMatrix) project type which is currently implemented in Visual Studio 2012. Using Framework 4.5 is not mandatory however is absolutely normal.
The SignalR Hubs API allows to process remote procedure calls (RPCs) from a server to connected clients and from clients to the server side. Add new class of type SignalR Hub Class to the existing the PubSub__NSB_SignalR.SignalRSubscriber1 project. If such template is absent in your VS2012 installation, add simply a new class instead. Add the following code to Chat.cs :
using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using Microsoft.AspNet.SignalR;
namespace PubSub__NSB_SignalR.SignalRSubscriber1
{
public class Chat : Hub
{
public void Send(string message)
{
Clients.All.send(message);
}
}
}
Add another new class RegisterHubs.cs into PubSub__NSB_SignalR.SignalRSubscriber1\App_Start folder, and add the following code:
using System.Web;
using System.Web.Routing;
using Microsoft.AspNet.SignalR;
[assembly: PreApplicationStartMethod(typeof(PubSub__NSB_SignalR.SignalRSubscriber1.RegisterHubs), "Start")]
namespace PubSub__NSB_SignalR.SignalRSubscriber1
{
public class RegisterHubs
{
public static void Start()
{
var config = new HubConfiguration
{
EnableCrossDomain = true
};
RouteTable.Routes.MapHubs(config);
}
}
}
Change existing reference on Microsoft.AspNet.Signal.Core.dll ver. 1.0.0 added by NuGet on the ver.1.0.1. It's required because in NuGet's version EnableCrossDomain is undefined.
Next, add the following script directly to the end of PubSub__NSB_SignalR.SignalRSubscriber1\Views\Home\Index.cshtml file:
@section scripts {
<script src="Scripts/jquery-1.8.3.min.js"></script>
<script src="Scripts/jquery.signalR-1.0.0.min.js"></script>
<script src="http://localhost:19897/signalR/hubs"></script>
<script>
$(function () {
var chat = $.connection.chat;
chat.client.send = function (message) {
$('#message').append('<li>' + message + '</li>');
};
$.connection.hub.logging = true;
$.connection.hub.url = 'http://localhost:19897/signalr';
$.connection.hub.start().done(function () {
$('#send').click(function () {
chat.server.send($('#msg').val());
});
});
});
</script>
}
<input type="text" id="msg" value=" " />
<input type="button" id="send" value="send" />
<ul id="message">
</ul>
Verify that the project has reference to Scripts/jquery-1.8.3.min.js and Scripts/jquery.signalR-1.0.0.min.js.
Compile and launch the PubSub__NSB_SignalR.SignalRSubscriber1 project separately (right click Debug | Start new instance). In this case it is on localhost:19897. Let's remember this port because it will be referenced to.
Verify that the following screen is received, if so, done well. This screen indicates that the Send method of Chat.cs class is called.
Now let's add to PubSub__NSB_SignalR solution two new projects BasicChat.Mvc and BasicChat.Wpf from an external source: https://github.com/CasperWollesen/SignalR.Samples. Few small mandatory changes are required:
<a name="ch1" id="A14">Change #1:</a> In BasicChat.Mvc\Global.asax.cs remove RoubteTable.Routes.MapHubs(); because it was called already in the file : PubSub__NSB_SignalR.SignalRSubscriber1\App_Start\RegisterHubs.cs.
using System.Web;
using System.Web.Http;
using System.Web.Mvc;
using System.Web.Optimization;
using System.Web.Routing;
using Microsoft.AspNet.SignalR;
namespace BasicChat.Mvc
{
public class MvcApplication : System.Web.HttpApplication
{
protected void Application_Start()
{
AreaRegistration.RegisterAllAreas();
WebApiConfig.Register(GlobalConfiguration.Configuration);
FilterConfig.RegisterGlobalFilters(GlobalFilters.Filters);
RouteConfig.RegisterRoutes(RouteTable.Routes);
BundleConfig.RegisterBundles(BundleTable.Bundles);
AuthConfig.RegisterAuth();
}
}
}
<a name="ch2" id="A15">Change #2:</a> In BasicChat.Mvc\Views\Home\Index.cshtml replace the script with the following:
@section scripts {
<script src="Scripts/jquery-1.8.3.min.js"></script>
<script src="Scripts/jquery.signalR-1.0.0.min.js"></script>
<script src="http://localhost:19897/signalR/hubs"></script>
<script>
$(function () {
var chat = $.connection.chat;
chat.client.send = function (message) {
$('#message').append('<li>' + message + '</li>');
};
$.connection.hub.logging = true;
$.connection.hub.url = 'http://localhost:19897/signalr';
$.connection.hub.start().done(function () {
$('#send').click(function () {
chat.server.send($('#msg').val());
});
});
});
</script>
}
In BasicChat.Wpf project open the following files:
SignalrClientSeparatedConnectionProxyWindow.xaml.cs,-
SignalrClientSharedConnectionProxyWindow.xaml.cs.
Don't forget to change port (in this example's case on 19897). In order to switch between them enter corresponding change into Application.StartupUri attribute in BasicChat.Wpf\App.xaml.
SignalrClientSeparatedConnectionProxyWindow.cs
using System;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Input;
using System.Windows.Threading;
using Microsoft.AspNet.SignalR.Client.Hubs;
namespace BasicChat.Wpf
{
public partial class <code>SignalrClientSeparatedConnectionProxyWindow</code> : Window
{
public System.Threading.Thread Thread { get; set; }
public string Host = "http://localhost:19897/";
public IHubProxy ProxyInvoke { get; set; }
public HubConnection ConnectionInvoke { get; set; }
public bool Active { get; set; }
public SignalrClientSeparatedConnectionProxyWindow()
{
InitializeComponent();
}
private async void ActionSendButtonClick(object sender, RoutedEventArgs e)
{
await SendMessage();
}
private async Task SendMessage()
{
await ProxyInvoke.Invoke("send", ClientNameTextBox.Text + ": " + MessageTextBox.Text);
}
private async void ActionWindowLoaded(object sender, RoutedEventArgs e)
{
Active = true;
Thread = new System.Threading.Thread(() =>
{
var connectionOn = new HubConnection(Host);
IHubProxy proxyOn = connectionOn.CreateHubProxy("Chat");
Proxy.On<string>("send", OnSendData);
connectionOn.Start();
while (Active)
{
System.Threading.Thread.Sleep(10);
}
}) { IsBackground = true };
Thread.Start();
ConnectionInvoke = new HubConnection(Host);
ProxyInvoke = ConnectionInvoke.CreateHubProxy("Chat");
await ConnectionInvoke.Start();
}
private void OnSendData(string message)
{
Dispatcher.Invoke(DispatcherPriority.Normal, (Action)(() => MessagesListBox.Items.Insert(0, message)));
}
private async void ActionMessageTextBoxOnKeyDown(object sender, KeyEventArgs e)
{
if (e.Key == Key.Enter || e.Key == Key.Return)
{
await SendMessage();
MessageTextBox.Text = "";
}
}
}
}
SignalrClientSharedConnectionProxyWindow.cs
using System;
using System.Threading.Tasks;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Input;
using System.Windows.Threading;
using Microsoft.AspNet.SignalR.Client.Hubs;
namespace BasicChat.Wpf
{
public partial class <code>SignalrClientSharedConnectionProxyWindow</code> : Window
{
public System.Threading.Thread Thread { get; set; }
public string Host = "http://localhost:19897/";
public IHubProxy Proxy { get; set; }
public HubConnection Connection { get; set; }
public bool Active { get; set; }
public SignalrClientSharedConnectionProxyWindow()
{
InitializeComponent();
}
private async void ActionSendButtonClick(object sender, RoutedEventArgs e)
{
await SendMessage();
}
private async Task SendMessage()
{
await Proxy.Invoke("send", ClientNameTextBox.Text + ": " + MessageTextBox.Text);
}
private void ActionWindowLoaded(object sender, RoutedEventArgs e)
{
Active = true;
Thread = new System.Threading.Thread(() =>
{
Connection = new HubConnection(Host);
Proxy = Connection.CreateHubProxy("Chat");
Proxy.On<string>("send", OnSendData);
Connection.Start();
while (Active)
{
System.Threading.Thread.Sleep(10);
}
}) { IsBackground = true };
Thread.Start();
}
private void OnSendData(string message)
{
Dispatcher.Invoke(DispatcherPriority.Normal, (Action)(() => MessagesListBox.Items.Insert(0, message)));
}
private async void ActionMessageTextBoxOnKeyDown(object sender, KeyEventArgs e)
{
if (e.Key == Key.Enter || e.Key == Key.Return)
{
await SendMessage();
MessageTextBox.Text = "";
}
}
}
}
In order to define methods on the client that a Hub can call from the server creates a proxy for the Hub by calling CreateHubProxy on the connection object. When the proxy is created you can invoke method calls on the remote hub and pass parameters. SignalR provides this functionality with the use of IHubProxy.
Remark: In order to create a console application, the main function will be very similar to the WPF variant shown above.
Now, having the two previously tested parts: NSB and SignalR it's time to assemble them together. In
PubSub__NSB_SignalR\PubSub__NSB_SignalR.Code\Notifications\EventMessageProcessor.cs from the NSB part, add the changes as shown below:
using System;
using NServiceBus;
using PubSub__NSB_SignalR.Contract.Notifications;
using Microsoft.AspNet.SignalR;
using Microsoft.AspNet.SignalR.Hubs;
using PubSub__NSB_SignalR.SignalRSubscriber1;
namespace PubSub__NSB_SignalR.Notifications
{
public partial class EventMessageProcessor
{
partial void HandleImplementation(EventMessage message)
{
System.Diagnostics.Trace.TraceInformation("Notifications received: {0}", message.GetType().Name);
IHubContext hubContext = GlobalHost.ConnectionManager.GetHubContext<Chat>();
hubContext.Clients.All.send(message.Time.ToString());
}
}
}
Suppose timer notification should be sent every 3 seconds by the publisher to the subscriber via NServiceBus, in PubSub__NSB_SignalR\PubSub__NSB_SignalR.MyPublisher\ServerEndpoint.cs add the changes as shown below:
using System;
using NServiceBus;
using PubSub__NSB_SignalR.Contract.Notifications;
namespace PubSub__NSB_SignalR.MyPublisher
{
class ServerEndpoint : IWantToRunWhenBusStartsAndStops
{
public IBus Bus { get; set; }
public void Start()
{
Console.WriteLine("This will publish IEvent, EventMessage, and AnotherEventMessage alternately.");
Console.WriteLine("Press 'Enter' to publish a timer message.To exit, Ctrl + C");
while (Console.ReadLine() != null)
{
var timer = new System.Timers.Timer(1000);
timer.Elapsed += timer_Elapsed;
timer.Interval = 3000;
timer.Enabled = true;
EventMessage eventMessage = new EventMessage();
eventMessage.EventId = Guid.NewGuid();
eventMessage.Time = DateTime.Now;
eventMessage.Duration = TimeSpan.FromSeconds(99999D);
Bus.Publish(eventMessage);
Console.WriteLine("Published event with Id {0} {1}", eventMessage.EventId, eventMessage.Time.ToString());
Console.WriteLine("==========================================================================");
}
}
public void Stop()
{
}
void timer_Elapsed(object sender, System.Timers.ElapsedEventArgs e)
{
EventMessage eventMessage = new EventMessage();
eventMessage.EventId = Guid.NewGuid();
eventMessage.Time = DateTime.Now;
eventMessage.Duration = TimeSpan.FromSeconds(99999D);
Bus.Publish(eventMessage);
Console.WriteLine("Published event with Id {0} {1}", eventMessage.EventId, eventMessage.Time.ToString());
Console.WriteLine("==========================================================================");
}
}
}
Open the Package Manager Console via Tools | Library Packages Manager and fetch through NuGet the SignalR development packages:
PM> Install-Package Microsoft.AspNet.SignalR.Client -Version 1.0.1 –ProjectName BasicChat.Wpf
Build the BasicChat.Wpf project.
In such a way, whenever subscribed event is handled (in this case the EventMessage), the message is resent via SignalR to our client's applications: BasicChat.Mvc and BasicChat.Wpf using our Chat-hub class. By add into PubSub__NSB_SignalR.SignalRSubscriber1\Controllers\HomeController.cs the changes as shown below, this will allow to send messages from PubSub__NSB_SignalR.SignalRSubscriber1 to BasicChat.Mvc and BasicChat.Wpf.
public ActionResult About()
{
ViewBag.Message = "Your app description page.";
IHubContext hubContext = GlobalHost.ConnectionManager.GetHubContext<chat>();
hubContext.Clients.All.send("XXXX");
return View();
}
<a name="lo" id="A24">Launching order</a>: First run NServiceBus publisher – subscriber pair (press F5 in order to Debug the solution) and only then run one-by-one the clients (Debug | Start new instance): BasicChat.Mvc and BasicChat.Wpf.If all done well, following results are received:
Remark: The arrows represent the possible communication channels.
Summary
One complete cycle of communication between publisher and subscriber was shown, demonstrating proposed approach which consists of consistent use of NServiceBus (Service Matrix) with SignalR for the aim of CQRS architecture implementation.
