Using SignalR To Push StreamInsight Events to Client Browsers

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

//added reference to SignalR
using SignalR.Hubs;

///
<summary> /// Summary description for BizEventController /// </summary>

public class BizEventController : Hub
{
    public void AddSubscription(string eventType)
    {
        AddToGroup(eventType);
    }
}

$(function () {
            //create arrays for use in showing formatted date string
            var days = ['Sun', 'Mon', 'Tues', 'Wed', 'Thur', 'Fri', 'Sat'];
            var months = ['Jan', 'Feb', 'Mar', 'Apr', 'May', 'June', 'July', 'Aug', 'Sept', 'Oct', 'Nov', 'Dec'];

            // create proxy that uses in dynamic signalr/hubs file
            var bizEDeck = $.connection.bizEventController;

            // Declare a function on the chat hub so the server can invoke it
            bizEDeck.addEventMsg = function (message) {
                //format date
                var receiptDate = new Date();
                var formattedDt = days[receiptDate.getDay()] + ' ' + months[receiptDate.getMonth()] + ' ' + receiptDate.getDate() + ' ' + receiptDate.getHours() + ':' + receiptDate.getMinutes();
                //add new "message" to deck column
                $('#deck').prepend('</pre>
<div>' + message + '' + formattedDt + ' via StreamInsight</div>
<pre>
');
            };

            //act on "subscribe" button
            $("#groupadd").click(function () {
                //call subscription function in server code
                bizEDeck.addSubscription($('#group').val());
                //add entry in "subscriptions" section
                $('#subs').append($('#group').val() + '</pre>

<hr />

<pre>');
            });

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

using System;
using System.Collections.Generic;
using System.Linq;
using System.Runtime.Serialization;
using System.ServiceModel;
using System.Text;

using SignalR;
using SignalR.Infrastructure;
using SignalR.Hosting.AspNet;
using StreamInsight.Samples.Adapters.Wcf;
using Seroter.SI.AzureAppFabricAdapter;

public class NotificationService : IPointEventReceiver
{
	//implement the operation included in interface definition
	public ResultCode PublishEvent(WcfPointEvent result)
	{
		//get category from key/value payload
		string cat = result.Payload["Category"].ToString();
		//get message from key/value payload
		string msg = result.Payload["EventMessage"].ToString();

		//get SignalR connection manager
		IConnectionManager mgr = AspNetHost.DependencyResolver.Resolve();
		//retrieve list of all connected clients
		dynamic clients = mgr.GetClients();

		//send message to all clients for given category
		clients[cat].addEventMsg(msg);
		//also send message to anyone subscribed to all events
		clients["All"].addEventMsg(msg);

		return ResultCode.Success;
	}
}

using System;
using System.Collections.Generic;
using System.Linq;
using System.Text;

using Microsoft.ComplexEventProcessing;
using Microsoft.ComplexEventProcessing.Linq;
using Seroter.SI.AzureAppFabricAdapter;
using StreamInsight.Samples.Adapters.Wcf;

namespace SignalRTest.StreamInsightHost
{
    class Program
    {
        static void Main(string[] args)
        {
            Console.WriteLine(":: Starting embedded StreamInsight server ::");

            //create SI server
            using(Server server = Server.Create("RSEROTERv12"))
            {
                //create SI application
                Application app = server.CreateApplication("SeroterSignalR");

                //create input adapter configuration
                WcfAdapterConfig inConfig = new WcfAdapterConfig()
                {
                    Password = "",
                    RequireAccessToken = false,
                    Username  = "",
                    ServiceAddress = "http://localhost:80/StreamInsightv12/RSEROTER/InputAdapter"
                };

                //create output adapter configuration
                WcfAdapterConfig outConfig = new WcfAdapterConfig()
                {
                    Password = "",
                    RequireAccessToken = false,
                    Username = "",
                    ServiceAddress = "http://localhost:6412/SignalRTest/NotificationService.svc"
                };

                //create event stream from the source adapter
                CepStream input = CepStream.Create("BizEventStream", typeof(WcfInputAdapterFactory), inConfig, EventShape.Point);
                //build initial LINQ query that is a simple passthrough
                var eventQuery = from i in input
                                 select i;

                //create unbounded SI query that doesn't emit to specific adapter
                var query0 = eventQuery.ToQuery(app, "BizQueryRaw", string.Empty, EventShape.Point, StreamEventOrder.FullyOrdered);
                query0.Start();

                //create another query that latches onto previous query
                //filters out all individual web hits used in later agg query
                var eventQuery1 = from i in query0.ToStream()
                                  where i.Category != "Web"
                                  select i;

                //another query that groups events by type; used here for web site hits
                var eventQuery2 = from i in query0.ToStream()
                                  group i by i.Category into EventGroup
                                  from win in EventGroup.TumblingWindow(TimeSpan.FromSeconds(10))
                                  select new BizEvent
                                  {
                                      Category = EventGroup.Key,
                                      EventMessage = win.Count().ToString() + " web visits in the past 10 seconds"
                                  };
                //new query that takes result of previous and just emits web groups
                var eventQuery3 = from i in eventQuery2
                                  where i.Category == "Web"
                                  select i;

                //create new SI queries bound to WCF output adapter
                var query1 = eventQuery1.ToQuery(app, "BizQuery1", string.Empty, typeof(WcfOutputAdapterFactory), outConfig, EventShape.Point, StreamEventOrder.FullyOrdered);
                var query2 = eventQuery3.ToQuery(app, "BizQuery2", string.Empty, typeof(WcfOutputAdapterFactory), outConfig, EventShape.Point, StreamEventOrder.FullyOrdered);

                //start queries
                query1.Start();
                query2.Start();
                Console.WriteLine("Query started. Press [Enter] to stop.");

                Console.ReadLine();
                //stop all queries
                query1.Stop();
                query2.Stop();
                query0.Stop();
                Console.Write("Query stopped.");
                Console.ReadLine();

            }
        }

        private class BizEvent
        {
            public string Category { get; set; }
            public string EventMessage { get; set; }
        }
    }
}