(untagged)

Streaming live results to a web site using MSMQ/Duplex WCF/SignalR/jQuery

Richard E King

0.00/5 (No votes)

7 Feb 2012

Streaming live results to a web site using MSMQ/Duplex WCF/SignalR/jQuery

Demo project : Codeproject.EventBroker.zip

Introduction
Video
How To Run The Demo
PreRequisites
General Design
That's it

Introduction

Setting the scene......Where I work is a Fx (Foreign Exchange) company and we trade Fx for clients all over the world, and the other day my boss came up to me and stated that he would like to be able to visualise where trades where they were happening in real time, but he wanted it too look cool, a kind of shiny showcase type of thing (I am sure you know what I mean). He categorically stated no grids. I was pleased.

My team mate Richard and I were tasked with this, so we thought about this and looked at what sort of information we had available, and wondered if we could make some sort of generic real time event watcher that would also produce some sparkly interface for us to show off.

We did NOT have much information to work with, we pretty much ONLY had the following

Tcpip address
An arbitary string that described the real time event type, for example "ClientDeal", "ExchangeFund" etc etc
The name of the client

So we thought about it a bit more, and what we came up with was something along the lines of this scenario.

We could extend our logging framework (we use Log4Net), where we could create a custom MessageQueue (MSMQ) appender, which would log certain events and some extra data (such as Tcpip address) to a MessageQueue. Obviously we could not share our entire application so we have provider a test message publisher that simply writes test messages to a MessageQueue. This part should be pretty easy to figure out should you want to come up with your own stuff to generate the real time events.
We could have a WCF service read these MessageQueue entries in real time. This service could take in subscribers, where each subscriber could subscribe for a single event using the event name, or multiple events, by passing an array of event names on subscription
We could make the WCF service use callbacks to the subscribers to push notifications back to the subscribers in real time
We could also use Google Earth to show these events (if we could obtain GeoLocation data for the event) in real time

Most of this is fairly standard stuff, the actual interesting part is pushing notifications from some server side web code back to the browser in real time. I don't know how many of you have seen that before, but it is like what Google do when you open a Google search page and search for something that is quite populaar (say some news worthy item), and Google will actual stream live results at you straight into your open search page.

It's very cool, and is usually accomplished using long polling or various other techniques using Ajax/Comet techniques, all of which are very hard to set up and get working (at least in our opinion).

What we ended up doing was using a rather new library called SignalR, which I have say is pretty darned cool.

We will be going through all of the different moving parts of how we went about this, later in the article. One thing to bear in mind is that for our requirement we wanted to show stuff on Google Earth, which you may not want to do. However the use of SignalR could be applied to any scenario where you want to stream live data straight to users browsers, something like search results, some sort of streaming changing data, such as live market rates, or strangely enough FX rates. Funny that.

Demo Video

This demo video shows the web project receiving events in real time from the test publisher project. The full path of what is happening is this:

Msmq -> WCF -> Xml Parsing -> GeoLocation lookup -> WCF Callback -> Website -> SignalR -> Javascript -> Google Earth API

One thing to note is that the test publisher is picking randomly from a small set of known TcpIp addresses, so you may see the same TcpIp (therefor GeoLocation) picked after each other. That is the nature of randomness over a small dataset.

Anyway click the image below to download the video (its about 180Mb sorry, screen capture software produces big files), it should be pretty clear what is happening, essentially the test publisher project is publishing event to an MessageQueue and the these messages are being pushed to Google Earth in real time (streamed to it) via WCF and the use of the SignalR library.

How To Run The Demo

The demo code actually comprises a few projects that must be run in a particular order, so here is what needs to be done to run the demo code correctly:

Run the Codeproject.EventBroker.TestMessagePublisher project and choose Automatic mode
Run the Codeproject.EventBroker.Host project (do it in DEBUG as it will be self hosted console WCF app then)
Run the Codeproject.EventBroker.WebUI and wait a little while and you should see Google Earth navigating to different parts of the world

PreRequisites

There are a few prerequisites, however most of them are included as part of the attached demo code. The table below shows them all and tells you whether they are included in the attached demo code, or whether you really MUST have them in order to run the code

Item	Included
MSMQ	NO You MUST have this turned on and running (it is a standard Windows component)
Named MSMQ Private Queue	NO You MUST have a queue named "eventbroker" (or whatever you configure in the TestMessagePublisher and WebUI project config files)
IIS Express 7.5	NO You MUST install this from Microsofts download page : http://www.microsoft.com/download/en/details.aspx?id=1038
Castle Windsor	YES See Lib\Castle\1.2.0.6623 folder
Log4Net	YES See Lib\Log4Net\1.2.10.0\log4net.dll
SignalR	YES See Lib\Microsoft\SignalR\SignalR.dll

General Design

I think the best way to get started is to consider the following diagram which tries to outline the different parts of the attached demo code:

Each of the black outlined boxes represents a project within the demo project, whilst the orange outlined box represents functionality that is exposed by the use of the SignalR dll.

We will be going into these projects and the use of SignalR in more detail below, but for now here is a very short description of what each of these projects does.

Codeproject.EventBroker.TestMessagePublisher: This is throw away code. The sole job of this project is to simulate the generation of real time messages
Codeproject.EventBroker.Service: This is a duplex WCF service that reads messages from a MessageQueue that the TestMessagePublisheris writing to. To run this service you will need to start the WCF host project Codeproject.EventBroker.Host
Codeproject.EventBroker.WebUI: This is stndard ASP .NET project which host an instance of Google Earth in a web page. The web page also calls some server side SignalR code which then subscribes to the WCF service and will also accept callbacks which provide real time values which are then shown in real time on Google Earth using SignalR.

Test Publisher

As we have already stated the Codeproject.EventBroker.TestMessagePublisherproject is a throw away peice of code that is simply used to simulate real time messages occurring.

When you run this project it will look something like this.

It can seen that there are 2 radio buttons and a start button.The 2 radio buttons are used to determine how test messages are written to the MessageQueue.

Automatic : A new message is created every x-seconds after clicking start
Manual: A new message is ONLY created when you click start

Here is the relevant code that does the writing to the MessageQueue

using System;
using System.Collections.Generic;
using System.ComponentModel;
using System.Configuration;
using System.Data;
using System.Diagnostics;
using System.Drawing;
using System.Linq;
using System.Messaging;
using System.Text;
using System.Threading;
using System.Windows.Forms;
using System.Threading.Tasks;
using Message = System.Windows.Forms.Message;




namespace Codeproject.EventBroker.TestMessagePublisher
{
    public partial class MainWindow : Form
    {
        private enum RunMode { Automatic = 1, Manual }
        private RunMode CurrentRunMode = RunMode.Automatic;
        private string inputQueueName = ConfigurationManager.AppSettings["eventBrokerQueueName"];
        private List<string> places = new List<string>();
        private List<int> waits = new List<int>();
        private Random rand = new Random();
        private bool listenToSelectionChanges = true;
        private bool stopAuto = false;

        public MainWindow()
        {
            InitializeComponent();
            places.Add("220.233.19.142");
            places.Add("64.233.160.0");
            places.Add("91.135.229.5");

            waits.Add(1000);
            waits.Add(2000);
            waits.Add(4000);
            waits.Add(5000);
            waits.Add(8000);
        }

        public void SendMessages()
        {
            Task.Factory.StartNew(() =>
            {
                while (!stopAuto)
                {
                    SendMessage();
                    Thread.Sleep(10000);
                }
            }, TaskCreationOptions.LongRunning);
        }
        
        public string GetXmlData(string tcpIpAddress)
        {
            return string.Format(
                "<realtimeEvent>" +
                  "<originatingIp>{0}</originatingIp>" +
                  "<eventName>ClientDealEvent</eventName>" +
                  "<entityIdType>ClientDeal</entityIdType>" +
                  "<description>Someone bought something</description>" +
                  "<date>{1}</date>" +
                  "<additionalData></additionalData>" +
                "</realtimeEvent>", tcpIpAddress, DateTime.Now);
        }

        private void btnCreateManual_Click(object sender, EventArgs e)
        {
            if (radAuto.Checked)
            {
                SendMessages();
            }
            else
            {
                SendMessage();
            }
        }

        private void SendMessage()
        {
            using (MessageQueue queue = new MessageQueue(inputQueueName, QueueAccessMode.Send))
            {
                queue.Formatter = new XmlMessageFormatter(new[] { typeof(string) });
                try
                {
                    System.Messaging.Message message = new System.Messaging.Message(
                        GetXmlData(places[rand.Next(places.Count)]));
                    Debug.WriteLine("Producing message {0}", message.Body.ToString());
                    queue.Send(message);
                }
                catch (MessageQueueException mex)
                {
                    if (mex.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                    {
                        Debug.WriteLine("Message queue exception occured", mex);
                    }
                }
                catch (Exception ex)
                {
                    // Write the message details to the Error queue
                    Debug.WriteLine("Exception occured", ex);
                }
            }
        }

        private void CheckedChanged(object sender, EventArgs e)
        {
            if (radAuto.Checked)
            {
                stopAuto = false;
            }
            else
            {
                stopAuto = true;
            }
        }
    }
}

It can be seen that the message structure we went for is a small bit of XML that looks something like this

<realtimeEvent>
   <originatingIp>192.168.0.1</originatingIp>
   <eventName>ClientDealEvent</eventName>
   <entityIdType>ClientDeal</entityIdType>
   <description>Someone bought something</description>
   <date>02/01/2012</date>
   <additionalData></additionalData>
</realtimeEvent>

Where we configure what MessageQueue queue to use is done within the App.Config of the Codeproject.EventBroker.TestMessagePublisherproject

<?xml version="1.0"?>
<configuration>
  <appSettings>
    <add key="eventBrokerQueueName" value="FormatName:Direct=OS:localhost\private$\eventbroker"/>
  </appSettings>
  <startup>
    <supportedRuntime version="v4.0" sku=".NETFramework,Version=v4.0"/>
  </startup>
</configuration>

It can be seen that the default queue name is "eventbroker" which is expected to be a private queue on the local machine. But this queue could be anywhere, this is just to show you where you can configure this.

One other important note is that the "eventbroker" MessageQueue queue MUST NOT be transactional. As the demo code assumes it is not transactional, if you want to make the queue transactional you will need to alter the Codeproject.EventBroker.TestMessagePublisher MessageQueue writing code and the Codeproject.EventBroker.Service MessageQueue reading code.

If you want to make it transactional that's fine, but you WILL have to change code to do that. Also be aware that you will have to give access rights to a user to the newly created "eventbroker" MessageQueue queue. I tend to go with my own login and grant all rights.

Duplex WCF Service

The Codeproject.EventBroker.Service is a Duplex WCF service that basically has the following contract that the client may call

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using Codeproject.EventBroker.Contracts.Faults;

namespace Codeproject.EventBroker.Contracts.Service
{
    [ServiceContract(Namespace = "http://Codeproject.EventBroker.Contracts", 
                     SessionMode=SessionMode.Required,
                     CallbackContract=typeof(IEventBrokerCallback))]
    public interface IEventBroker
    {
        [OperationContract(IsOneWay = false)]
        [FaultContract(typeof(EventBrokerException))]
        void Subscribe(Guid subscriptionId, string[] eventNames);

        [OperationContract(IsOneWay = true)]
        void EndSubscription(Guid subscriptionId);
    }
}

This service also expects client to supply a callback contract that satisfies this interface

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using Codeproject.EventBroker.Contracts.Data;

namespace Codeproject.EventBroker.Contracts.Service
{
    public interface IEventBrokerCallback
    {
        [OperationContract(IsOneWay = true)]
        void ReceiveStreamingResult(RealTimeEventMessage streamingResult);
    }
}

This service is hosted in the Codeproject.EventBroker.Host project, which when run in RELEASE mode will be a Windows service host for the Codeproject.EventBroker.Contracts.Service.EventBroker,and when run in DEBUG mode will be a simple console app that hosts the Codeproject.EventBroker.Contracts.Service.EventBroker WCF service.

Where the Codeproject.EventBroker.Contracts.Service.EventBrokerserviceskeleton implementation looks like this

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using Codeproject.EventBroker.Contracts.Service;
using Codeproject.EventBroker.Service.Data;
using System.Threading.Tasks;
using System.Configuration;
using Codeproject.EventBroker.Contracts.Faults;
using System.Messaging;
using Codeproject.EventBroker.Common;
using Codeproject.EventBroker.Service.Utils;
using Codeproject.EventBroker.Contracts.Data;
using Codeproject.EventBroker.Service.Extensions;
using Codeproject.EventBroker.Service.Services.Contracts;
using System.Threading;

namespace Codeproject.EventBroker.Service
{
    [ServiceBehavior(    InstanceContextMode = InstanceContextMode.Single, 
            ConcurrencyMode = ConcurrencyMode.Multiple)]
    public class EventBroker : IEventBroker
    {
        ....
        ....
        ....

        public EventBroker()
        {
            inputQueueName = ConfigurationManager.AppSettings["eventBrokerQueueName"].ToString();
            StartCollectingMessage();
            xmlParser = IOCManager.Instance.Container.Resolve<IXmlParser>();
           }

        public void StartCollectingMessage()
        {
         ....
         ....
         ....
        }

        public void Subscribe(Guid subscriptionId, string[] eventNames)
        {
         ....
         ....
         ....
        }

        public void EndSubscription(Guid subscriptionId)
        {
         ....
         ....
         ....
        }
    }
}

Accepting Subscribers

When new subscriber subscribes to this WCF the following occurs

For each event name that the subscribers wishes to subscribe for do the following:
1. If there is currently no subscriptions for the that event name, created an empty subscription list
2. See if there is already a subscription list for that event name, if there is add the subscribers Id and callback context (IEventBrokerCallback) to the global dictionary of subscribers for the event name

Here is the most relevant code for a subscription occurring:

private void CreateSubscription(Guid subscriptionId, string[] eventNames)
{
    //Ensure that a subscription is created for each message type the subscriber wants to receive
    lock (syncObj)
    {
        foreach (string eventName in eventNames)
        {
            if (!eventNameToCallbackLookups.ContainsKey(eventName))
            {
                List<UniqueCallbackHandle> currentCallbacks = new List<UniqueCallbackHandle>();
                eventNameToCallbackLookups[eventName] = currentCallbacks;
            }
            eventNameToCallbackLookups[eventName].Add(
                new UniqueCallbackHandle(subscriptionId, 
            OperationContext.Current.GetCallbackChannel<IEventBrokerCallback>()));
        }
    }
}

Removing Subscriptions

Once a sunscriber chooses to end their subscription they may do so using the void EndSubscription(Guid subscriptionId) operation contract.

When subscriber ends a subscription the following occurs

For each event name in the global dictionary of subscribers for the event name
1. Get all subcriptions that do NOT have the same subscriptionId as the subscriber that is unsubscribing
2. Create a new global dictionary of subscribers of those subscriptions that remain after removing all subscriptions that are no longer need due to a subscriber ending a subscription

Here is the most relevant code when a EndSubscription occurring:

public void EndSubscription(Guid subscriptionId)
{
    lock (syncObj)
    {
        //create new dictionary that will be populated by those remaining
        Dictionary<string, List<UniqueCallbackHandle>> remainingEventNameToCallbackLookups = 
            new Dictionary<string, List<UniqueCallbackHandle>>();

        foreach (KeyValuePair<string,List<UniqueCallbackHandle>>  kvp in eventNameToCallbackLookups)
        {
            //get all the remaining subscribers whos session id is not the same as the one we wish to remove
            List<UniqueCallbackHandle> remainingMessageSubscriptions = 
                kvp.Value.Where(x => x.CallbackSessionId != subscriptionId).ToList();
            if (remainingMessageSubscriptions.Any())
            {
                remainingEventNameToCallbackLookups.Add(kvp.Key, remainingMessageSubscriptions);
            }
        }
        //now left with only the subscribers that are subscribed
        eventNameToCallbackLookups = remainingEventNameToCallbackLookups;
    }
}

Doing Callbacks

The interesting part of the WCF service is the actual callback to the subscribers. Bu when should this callback occur?

Well the callback to a subscriber should only occur when we have something to deliver to them, which is when we have an incoming message from the MessageQueueand it matches a subscribers subscription desires (basically the incoming message EventName matches the subscribers EventNamethat they used when subscribing.

As this WCF service is intended to be used by many clients, there are several threads running, there is the main WCF thread, and there is also a new Thread spun up to handle reading from the MessageQueue and dispatching messages back to subscribers should the imcoming event EventName have active subscribers.

This process can pretty much be seen in the following two WCF methods

Read the incoming MessageQueue messages

private void GetMessageFromQueue()
{
    try
    {
        Task messageQueueReaderTask = Task.Factory.StartNew(() =>
            {
                using (MessageQueue queue = new MessageQueue(inputQueueName, QueueAccessMode.Receive))
                {
                    queue.Formatter = new XmlMessageFormatter(new[] { typeof(string) });

                    while (shouldRun)
                    {
                        Message message = null;
                        try
                        {
                            if (!queue.IsEmpty())
                            {
                                LogManager.Log.Debug("Receiving queue message");
                                message = queue.Receive(queueReadTimeOut);
                                ProcessMessage(message);
                            }
                        }
                        catch (MessageQueueException e)
                        {
                            if (e.MessageQueueErrorCode != MessageQueueErrorCode.IOTimeout)
                            {
                                LogManager.Log.Warn("Message queue exception occured", e);
                            }
                        }
                        catch (Exception e)
                        {
                            // Write the message details to the Error queue
                            LogManager.Log.Warn("Exception occured", e);
                        }
                    }
                }
            }, TaskCreationOptions.LongRunning);
    }
    catch (AggregateException ex)
    {
        throw;
    }
}

Do the callbacks to subscribers

The only other clever thing this code does, is if we get a CommunicationObjectAbortedException whilst trying to send a message to a subscriber, that subscriber is assumed to be faulted and removed. You will see that the subscriber also has mechanisms for dealing with a faulted channel, which is not so easy when it comes to publish/subscribe. For example one subscriber could be faulted but all others could be fine, so should we restart the ServiceHost, probably not. That is the appraoch we have taken here we try to be as fault tolerant as possible, and only resort to restarting the ServiceHost if the channel faults completely.

private void ProcessMessage(Message msmqMessage)
{
    string messageBody = (string)msmqMessage.Body;
    LogManager.Log.DebugFormat("ProcessMessage : {0}", messageBody);

    RealTimeEventMessage messageToSendToSubscribers = xmlParser.ParseRawMsmqXml(messageBody);
    if (messageToSendToSubscribers != null)
    {
        lock (syncObj)
        {
            List<Guid> deadSubscribers = new List<Guid>();
            if (eventNameToCallbackLookups.ContainsKey(messageToSendToSubscribers.EventName))
            {
                List<UniqueCallbackHandle> uniqueCallbackHandles = 
                    eventNameToCallbackLookups[messageToSendToSubscribers.EventName];
                foreach (UniqueCallbackHandle uniqueCallbackHandle in uniqueCallbackHandles)
                {
                    try
                    {
                        uniqueCallbackHandle.Callback.ReceiveStreamingResult(messageToSendToSubscribers);
                    }
                    catch(CommunicationObjectAbortedException coaex)
                    {
                        deadSubscribers.Add(uniqueCallbackHandle.CallbackSessionId);
                    }
                }
            }

            //end all subcriptions for dead subscribers
            foreach (Guid deadSubscriberId in deadSubscribers)
            {
                EndSubscription(deadSubscriberId);
            }
        }
    }
}

It can be seen that the code that processes the incoming MessageQueuemessage parses the xml transmitted message body into an actual RealTimeEventMessageobject prior to sending it to the subscribers. This xml parsing code is shown below

public class XmlParser : IXmlParser
{
    private IGeoLocator geoLocator;

    public XmlParser(IGeoLocator geoLocator)
    {
        this.geoLocator = geoLocator;
    }

    public RealTimeEventMessage ParseRawMsmqXml(string messageBody)
    {
        //<realtimeEvent>
        //  <originatingIp></originatingIp>
        //  <eventName>ClientDealEvent</eventName>
        //  <entityIdType>ClientDeal</entityIdType>
        //  <description>Someone bought something</description>
        //  <date>2012-01-16T15:31:31</date>
        //  <additionalData></additionalData>
        //</realtimeEvent>

        try
        {
            RealTimeEventMessage info = new RealTimeEventMessage();
            XElement xelement = XElement.Parse(messageBody);
            string ipAddress = GetSafeString(xelement, "originatingIp"); 
            if (!string.IsNullOrEmpty(ipAddress))
            {
                info.Location = geoLocator.ObtainLocationForIPAddress(ipAddress);
            }
            info.EventName = GetSafeString(xelement, "eventName");
            info.EntityIdType = GetSafeString(xelement, "entityIdType");
            info.Description = GetSafeString(xelement, "description").Replace("\n\n", "\n\r");
            info.Date = GetSafeDate(xelement, "date");
            info.AdditionalData = GetSafeString(xelement, "additionalData");
            return info;
        }
        catch (Exception ex)
        {
            LogManager.Log.Error(ex);
            return null;
        }
    }

    public static Int32 GetSafeInt32(XElement root, string elementName)
    {
        try
        {
            XElement element = root.Elements().Where(node => node.Name.LocalName == elementName).Single();
            return Convert.ToInt32(element.Value);
        }
        catch
        {
            return 0;
        }
    }

    private static DateTime? GetSafeDate(XElement root, string elementName)
    {
        try
        {
            XElement element = root.Elements().Where(node => node.Name.LocalName == elementName).Single();
            return DateTime.Parse(element.Value);
        }
        catch
        {
            return null;
        }
    }

    public static String GetSafeString(XElement root, string elementName)
    {
        try
        {
            XElement element = root.Elements().Where(node => node.Name.LocalName == elementName).Single();
            return element.Value;
        }
        catch
        {
            return String.Empty;
        }
    }       
}

Where this xml parsing code also make use of another bit of utility code that obtains GeoLocation information from a TcpIp address. This service is free but occassionally misses some TcpIp addresses. At work we actually use a web service provided by MindMap, which is very reliable and costs $20 for 10,000 lookups and is a simple GET http request. However for this articles demo code we have provided you with the free slightly unreliable version sorry abou that.

That said the TestMessageQueuePublisher is always picking random TcpIp addresses that we know work with the free GeoLocation lookup service that this demo code uses, so you should be fine.

Anyway here is the free (but slightly unreliable) GeoLocation lookup code:

public class GeoLocator : IGeoLocator
{
    public GeoLocation ObtainLocationForIPAddress(string ipAddress)
    {
        try
        {
            WebClient client = new WebClient();
            string locationDump = client.DownloadString(
                string.Format("http://api.hostip.info/get_html.php?ip={0}&position=true", 
                    ipAddress));
            string[] locationDumpSplit = locationDump.Split(
                new string[] { @"\r\n", @"\n" }, StringSplitOptions.RemoveEmptyEntries);

            decimal latitude = -1;
            decimal longitude = -1;
            int found=0;

            using (StringReader sr = new StringReader(locationDump)) 
            {
                found = 0;
                while (sr.Peek() >= 0) 
                {
                    string line = sr.ReadLine().ToLower();
                    if (line.StartsWith("latitude:"))
                    {
                        line = line.Replace("latitude:","").Trim();
                        latitude = decimal.Parse(line);
                        found++;
                    }
                    if (line.StartsWith("longitude:"))
                    {
                        line = line.Replace("longitude:", "").Trim();
                        longitude = decimal.Parse(line);
                        found++;
                    }
                }
            }

            if (found == 2)
            {
                return new GeoLocation(latitude, longitude);
            }
            else
                return null;
        }
        catch (Exception ex)
        {
            LogManager.Log.ErrorFormat(
                "Could not obtain Latitude/Longitude data for IpAddress {0}\r\n Exception : {1}",
                    ipAddress, ex);
            return null;
        }
    }
}

Web UI

The last peice to this puzzle is a simple web site that is used to display real time (or as near as damn it, there is a slight latency dealy go through the various layers, in fact these are the layers a real time event goes through, just so you can see where the web site fits it

Msmq -> WCF -> Xml Parsing -> GeoLocation lookup -> WCF Callback -> Website -> SignalR -> Javascript -> Google Earth API

Quite a path no!

Anyway that said the web site is pretty simply the only slightly exotic thing about it is that it uses this fairly new free library called SignalR which we discuss in greater details below. In essence what the web site does is host an instance of the Google Earth plugin in a standard HTML page which gets manipulated by a bit of jQuery Javascript. It also makes use of the SignalR library to allow push notifications to the browser.

SignalR Hub

Essentially what SignalR is, is a Async signaling library for ASP.NET to help build real-time, multi-user interactive web applications. It does this in a very clever way though. It basically allows you to write server side code that inherits from a SignalR Hub object. You can also then create Javascript that will communicate with the server side Hub object, and vice versa.

Yeah that's correct we can write to a Javascript method via server side code, it is actually quite nuts.

Of course there is some magic, but once you fathom what is going on, it is not that magical rather plain clever. Here is what happens under the covers

SignalR will create a lightweight Javascript proxy that allows Javascript to talk to the server side code
SignalR will also create dynamic “Clients” and “Caller” objects in your hub, so that you can invoke a client side method written in Javascript directly via your code in the server side
SignalR will examine your browser agents capabilities, and will do one of the following
- Will 1st attempt to use WebSockets to allow Javascript SignalR proxy to talk to server side code
- Failing the availability of WebSockets SignalR will revert to using long polling to allow Javascript SignalR proxy to talk to server side code

All of that is pretty hidden, so it is kind of magical.

There is a good SignalRa quickstart at the Hub Quickstart example which we suggest you read before we proceed. Once you have read that you will understand the code snippets below.

So for the demo project we have the following SignalR Hub

using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using SignalR.Hubs;




namespace Codeproject.EventBroker.WebUI.GeoLocation
{
    [HubName("eventTicker")]
    public class EventTickerHub : Hub
    {
        private int counter;
        private readonly EventTicker eventTicker;

        public EventTickerHub() : this(EventTicker.Instance) { }

        public EventTickerHub(EventTicker eventTicker)
        {
            this.eventTicker = eventTicker;
            eventTicker.Subscribe();
        }

        public void Register()
        {
            //Do nothing, but is crucially important to establish comms
        }
    }
}

Where the custom SignalR Hub also makes use of a EventTicker object. We will see more on that object later. For now that is all you need to know to create a custom SignalR Hub

Extra Material

Scott Hanselman has an excellent blog on SignalR at this link, it is well worth a read : http://www.hanselman.com/blog/AsynchronousScalableWebApplicationsWithRealtimePersistentLongrunningConnectionsWithSignalR.aspx

Code projects own Anoop Madhusudanan also just pipped us to the post by writing the 1st codeproject article on SignalR which is also worth a read: http://www.codeproject.com/Articles/322154/ASP-NET-MVC-SIngalR-and-Knockout-based-Real-time-U

Javascript Hub Comms

The JavaScript comms to the custom SignalR Hub is where the rest of the magic happens, but before we look at that, lets see what you need to do on a hosting page, HTML page in our case (could be ASP/ASPX/CSHTML etc etc)

It can be seen that we have the following script tags on the SignalR enabled page

<!DOCTYPE html PUBLIC "-//W3C//DTD XHTML 1.0 Transitional//EN" 
"http://www.w3.org/TR/xhtml1/DTD/xhtml1-transitional.dtd">
<html xmlns="http://www.w3.org/1999/xhtml">
<head>
    <script src="../Scripts/jquery-1.6.4.js"></script>
    <script src="../Scripts/jquery.signalR.js"></script>
    <script type="text/javascript" src="https://www.google.com/jsapi?key=ABCDEFG"> </script>
    <script src="../Scripts/jquery.color.js" type="text/javascript"></script>
    <script src="../signalr/hubs"></script>
    <script src="GeoLocationView.js"></script>
</head>
<body>
    <div id="map3d" style="height: 100%; width: 100%;">
    </div>
</body>
</html>

Now if you look at the demo projects folders, you will NOT see a signalr/hubs folder. That is magic, and you MUST just accept this and know that SignalR will be putting stuff there. Granted a certain element of trust is required.

So once you accept that there are unicorns/pixies and elfs in coding, we can now concentrate on reality which is how do we get JavaScript to talk to a SignalR Hub. In the demo code if you examine the file "GeoLocationView.js" you will see the following section of JavaScript code that is responsible for initiating the communications with the SignalR Hub.

var eventHub = $.connection.eventTicker;




//*************************************************
// Initialise SignalR Hub
//*************************************************
function InitialiseSignalRHub() {


    eventHub = $.connection.eventTicker;

    // Declare a function on the eventHub so the server can invoke it
    eventHub.addMessage = function (message) {
        ProcessGeoLocationCallbackMessage(message);
    }

    //callback that does nothing, simply here to establish link to Hub
    eventHub.registerCallback = function () {
    };


    // Start the connection
    $.connection.hub.start();

    //wait for 3s before we register with Hub 
    window.setTimeout(function () {
        eventHub.register();
    }, 3000)
}

There is also a callback done to the JavaScript from the SignalR Hub but we will see this in just a minute.

Subscribing To WCF Service

Subscribing to the duplex WCF sevice is a pretty standard affair, all we need to is something like this

using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using SignalR.Hubs;
using System.Timers;
using Codeproject.EventBroker.Contracts.Data;
using Codeproject.EventBroker.Contracts.Service;
using System.ServiceModel;
using Codeproject.EventBroker.WebUI.Wcf;
using System.Threading;
using System.ServiceModel.Channels;

namespace Codeproject.EventBroker.WebUI.GeoLocation
{
    public class EventTicker :  IEventBrokerCallback
    {
        private InstanceContext instanceContext;
        private Guid subscriptionId;
        EventBrokerProxy proxy;

        public EventTicker()
        {
            instanceContext = new InstanceContext(this);
            CreateProxy();
        }

        public void CreateProxy()
        {
            proxy = new EventBrokerProxy(instanceContext);
        }

        public void Subscribe()
        {
          ThreadPool.QueueUserWorkItem(x => 
            {
                try
                {
                    subscriptionId = Guid.NewGuid();
                    proxy.Subscribe(subscriptionId, 
            new string[] { "ClientDealEvent", "PaymentOutEvent" });
                    isSubscribed = true;
                }
                catch
                {
                }
            });
        }
    }
}

The only important parts of this are:

That we use a new InstanceContext to provide the WCF service with a callback object context.
That we MUST use a new thread to do the subscription. This is VERY important, as we need to keep the ASP .NET worker thread free, otherwise the callback will not work at all.

If you are curious here is the proxy code that the web site code uses to communicate with the duplex WCF service

using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;

namespace Codeproject.EventBroker.WebUI.Wcf
{
    public partial class EventBrokerProxy : 
        System.ServiceModel.DuplexClientBase<Codeproject.EventBroker.Contracts.Service.IEventBroker>,
        Codeproject.EventBroker.Contracts.Service.IEventBroker
    {
        public EventBrokerProxy(System.ServiceModel.InstanceContext callbackInstance) :
            base(callbackInstance)
        {
        }

        public EventBrokerProxy(System.ServiceModel.InstanceContext callbackInstance, 
            string endpointConfigurationName) :
            base(callbackInstance, endpointConfigurationName)
        {
        }

        public EventBrokerProxy(System.ServiceModel.InstanceContext callbackInstance, 
            string endpointConfigurationName, string remoteAddress) :
            base(callbackInstance, endpointConfigurationName, remoteAddress)
        {
        }

        public EventBrokerProxy(System.ServiceModel.InstanceContext callbackInstance, 
            string endpointConfigurationName, System.ServiceModel.EndpointAddress remoteAddress) :
            base(callbackInstance, endpointConfigurationName, remoteAddress)
        {
        }

        public EventBrokerProxy(System.ServiceModel.InstanceContext callbackInstance, 
            System.ServiceModel.Channels.Binding binding, System.ServiceModel.EndpointAddress remoteAddress) :
            base(callbackInstance, binding, remoteAddress)
        {
        }

        public void Subscribe(Guid subscriptionId, string[] eventNames)
        {
            base.Channel.Subscribe(subscriptionId, eventNames);
        }

        public void EndSubscription(Guid subscriptionId)
        {
            base.Channel.EndSubscription(subscriptionId);
        }
    }
}

Responding To A Callback

This is the really interesting part of this solution in our opinion, what happens is that when the duplex WCF service calls the EventTicker using the InstanceContext that the sunscriber provided, is that by using the SignalR Hub we are able to directly call into a Javascript method

Here is the relevant Codeproject.EventBroker.WebUI server side code (see EventTicker ), which is what the duplex WCF callback calls via the initial InstanceContext that was provided by the subscriber:

public void ReceiveStreamingResult(RealTimeEventMessage streamingResult)
{
    if (streamingResult.Location != null)
    {
        Hub.GetClients<EventTickerHub>().addMessage(streamingResult);
    }
}

Just as a reminder here is what the subscribers callback interface looks like:

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;
using System.ServiceModel;
using Codeproject.EventBroker.Contracts.Data;

namespace Codeproject.EventBroker.Contracts.Service
{
    public interface IEventBrokerCallback
    {
        [OperationContract(IsOneWay = true)]
        void ReceiveStreamingResult(RealTimeEventMessage streamingResult);
    }
}

And here is the relevant Codeproject.EventBroker.WebUI client side JavaScript code:

//*************************************************
// Initialise SignalR Hub
//*************************************************
function InitialiseSignalRHub() {

    ....
    ....
    ....
    // Declare a function on the eventHub so the server can invoke it
    eventHub.addMessage = function (message) {
        ProcessGeoLocationCallbackMessage(message);
    }
    ....
    ....
    ....
}




//*************************************************
// Process SignalR Hub callback
//*************************************************
function ProcessGeoLocationCallbackMessage(message) {

        //now add the items to the earth
        ShowPosition(message.Location.Latitude, message.Location.Longitude);
        CreateMarker(message.Location.Latitude, message.Location.Longitude, message.Description);
}

Pay special attention to the JavaScript function name, and see how the server side the SignalR Hub code is able to just call that, and pass across .NET objects which are then recieved by the client side JavaScript as JSON, that is quite mental we think. Quite mental indeed, Kudos to the SignalR chaps/chapesses.

As we have said before SignalR is clever enough to detect your browsers capabilities and will try the following

Start by attempting to use WebSockets (if they are supported)
Resort to using long polling if Web Sockets are not supported

Automatically UnSubscribing

One problem with do publish/subscribe is that the channel could fault for a particular subscriber and that channel for that subscriber and its callback is pretty much useless, but the subscriber has no way of knowing this. So how do we combat that.

Well if we look in the Web.Config of the Codeproject.EventBroker.WebUI project you will see the following WCF configuration

<system.serviceModel>

  <client>
    <endpoint name="Codeproject.EventBroker.Service.EventBroker"
              address="net.tcp://localhost:63747/EventBroker"
              binding="netTcpBinding"
              bindingConfiguration="DuplexBinding"
              contract="Codeproject.EventBroker.Contracts.Service.IEventBroker"/>
  </client>

  <bindings>
    <netTcpBinding>
      <binding name="DuplexBinding"
                sendTimeout="00:00:10"
                receiveTimeout="00:00:10">
        <reliableSession enabled="true"/>
        <security mode="None"/>
      </binding>
    </netTcpBinding>
  </bindings>

</system.serviceModel>

Where we see 2 timeout values Send and Receive, which are both set to 10 mins. So the approach we took was this, grab the Receive timeout value from the WCF binding, and the start a timer, when that timer expires we do an automatic unsubscibe to the WCF duplex service and then subscribe again. With this approach is place we only loose a maxium set of data for the Receive timeout value should the subscribers channel be faulted.

The most relevant code of the EventTicker is shown below.

using System;
using System.Collections.Generic;
using System.Linq;
using System.Web;
using SignalR.Hubs;
using System.Timers;
using Codeproject.EventBroker.Contracts.Data;
using Codeproject.EventBroker.Contracts.Service;
using System.ServiceModel;
using Codeproject.EventBroker.WebUI.Wcf;
using System.Threading;
using System.ServiceModel.Channels;

namespace Codeproject.EventBroker.WebUI.GeoLocation
{
    public class EventTicker :  IEventBrokerCallback
    {
        private InstanceContext instanceContext;
        private Guid subscriptionId;
        private bool isSubscribed;
        private TimeSpan receiveTimeout;
        private System.Timers.Timer subscriberLeaseRenewalTimer;
        EventBrokerProxy proxy;

        public EventTicker()
        {
            instanceContext = new InstanceContext(this);
            CreateProxy();
            Binding binding = proxy.Endpoint.Binding;
            receiveTimeout = binding.ReceiveTimeout;
            subscriberLeaseRenewalTimer = new System.Timers.Timer(receiveTimeout.TotalMilliseconds);
            subscriberLeaseRenewalTimer.Enabled = true;
            subscriberLeaseRenewalTimer.Start();
            subscriberLeaseRenewalTimer.Elapsed += SubscriberLeaseRenewalTimer_Elapsed;    
                
        }

        private void SubscriberLeaseRenewalTimer_Elapsed(object sender, ElapsedEventArgs e)
        {
            subscriberLeaseRenewalTimer.Enabled = false;
            subscriberLeaseRenewalTimer.Stop();
            EndSubscription();
            CreateProxy();
            Subscribe();
            subscriberLeaseRenewalTimer.Enabled = true;
            subscriberLeaseRenewalTimer.Start();
        }

        public void CreateProxy()
        {
            proxy = new EventBrokerProxy(instanceContext);
        }

        public void Subscribe()
        {
            ....
            ....
            ....
            ....
        }

        public void EndSubscription()
        {
            ....
            ....
            ....
            ....
        }
    }
}

Google Earth Integration

The Google Earth integration is all pretty standard stuff that you can learn by reading the various Google Earth documentation/API reference pages. However for completeness here is what the code looks like for Google Earth integration.

As we say this is all very standard stuff is you use the Google Earth API, essentially what we do is use the following code in the Codeproject.EventBroker.WebUI projects GeoLocationView.js file.

Initialise the Google Earth plugin
Create the Google Earth plugin navigation control
Navigates the Google Earth plugin to a particular Latitude/Longitude (that is provided by the callback to the SignalR Hub via the subscribers WCF callback context)
Shows a Google Earth plugin placemark for the current Latitude/Longitude (that is provided by the callback to the SignalR Hub via the subscribers WCF callback context)

//************************************************
// Global Vars
//************************************************

google.load('earth', '1');
var ge = null;
var placemark;

//************************************************
// Document Ready
//************************************************
$(function () {
    GlobalInit();
});


//*************************************************
// Global initialisation, hooks up Google Earth
// callback
//*************************************************
function GlobalInit() {
    google.setOnLoadCallback(EarthInit);
}

//*************************************************
// Initialise Googe Earth
//*************************************************
function EarthInit() {

    google.earth.createInstance(
        'map3d',
        function InitialisationPassed(instance) {
            ge = instance;
            console.log("InitialisationPassed " + ge);
            InitialiseSignalRHub();
            ge.getWindow().setVisibility(true);
            CreateNavigationControl();
        },
        function InitialisationFailed(errorCode) {
            console.log("InitialisationFailed " + errorCode);
            alert("there was an error initialising Google Earth\r\n" + errorCode);
        });
}


//*************************************************
// Creates Google Earth Navigation Control
//*************************************************
function CreateNavigationControl() {

    console.log("CreateNavigationControl " + ge);
    var geNavigationControl = ge.getNavigationControl();
    geNavigationControl.setVisibility(true);
    geNavigationControl.setStreetViewEnabled(true);
}

//*************************************************
// Navigates Google Earth To Particular Lat/Long
//*************************************************
function ShowPosition(lat, long) {
    // Get the current view
    var lookAt = ge.getView().copyAsLookAt(ge.ALTITUDE_RELATIVE_TO_GROUND);
    lookAt.setRange(lookAt.getRange() * 0.25);

    // Set new latitude and longitude values
    lookAt.setLatitude(lat);
    lookAt.setLongitude(long);

    // Update the view in Google Earth
    ge.getView().setAbstractView(lookAt);

    // Get the current view
    var camera = ge.getView().copyAsCamera(ge.ALTITUDE_RELATIVE_TO_GROUND);

    // Zoom out to twice the current distance
    camera.setAltitude(5000000);
    camera.setLatitude(lat);
    camera.setLongitude(long);

    // Update the view in Google Earth
    ge.getView().setAbstractView(camera);
}


//*************************************************
// Creates Google Earth Marker for Lat/Long 
// with description
//*************************************************
function CreateMarker(lat, long, desc) {

    //remove last placemark, only want to show 1 at a time
    if (placemark != undefined) {
        ge.getFeatures().removeChild(placemark);
    }

    placemark = ge.createPlacemark('');
    placemark.setName(desc);
    placemark.setDescription("Some cool stuff right here");

    // Define a custom icon.
    var icon = ge.createIcon('');

    var imageUrl = window.location.protocol + "//" + 
    window.location.host + "/content/Images/person.png";
    console.log(imageUrl);

    icon.setHref(imageUrl);
    var style = ge.createStyle(''); //create a new style
    style.getIconStyle().setIcon(icon); //apply the icon to the style
    style.getIconStyle().setScale(1.5);
    style.getLabelStyle().setScale(2.0);    
    placemark.setStyleSelector(style); //apply the style to the placemark

    // Set the placemark's location.  
    var point = ge.createPoint('');
    point.setLatitude(lat);
    point.setLongitude(long);
    placemark.setGeometry(point);

    // Add the placemark to Earth.
    ge.getFeatures().appendChild(placemark);
}

That's It

Anyway that is all we really wanted to say this time. Richard and I will probably have some more stuff for you at some stage, but I am now going to get back to the final 5% of this OSS project that Pete O'Hanlon and I are working on. Thing is that final 5% is the hardest part, but we are both into it, so expect to see it appearing here some time soon. Its been a while coming but we both like it, and feel it will be of use. So until then.....Dum dum dum.

However if you liked this article, and can be bothered to give a comment/vote they are always appreciated. Thanks for reading. Cheerio

License

This article has no explicit license attached to it but may contain usage terms in the article text or the download files themselves. If in doubt please contact the author via the discussion board below.

A list of licenses authors might use can be found here

Streaming live results to a web site using MSMQ/Duplex WCF/SignalR/jQuery

Table Of Contents

License