Author: Richard Seroter

In my previous post, I showed how to build StreamInsight adapters that receive Azure AppFabric messages and send Azure AppFabric messages.  In this post, we see how to use these adapters to push events into a cloud-hosted StreamInsight application and send events back out.

As a reminder, our final solution contains an on-premises consumer of an Azure AppFabric Service Bus endpoint and that event is relayed to StreamInsight Austin and output events are sent to an Azure AppFabric Service Bus endpoint that relays the event to an on-premises listener.

In order to follow along with this post, you would need to be part of the early adopter program for StreamInsight “Austin”. If not, no worries as you can at least see here how to build cloud-ready StreamInsight applications.

The StreamInsight “Austin” early adopter package contains a sample Visual Studio 2010 project which deploys an application to the cloud.  I reused the portions of that solution which provisioned cloud instances and pushed components to the cloud.  I changed that solution to use my own StreamInsight application components, but other than that, I made no significant changes to that project.

Let’s dig in.  First, I logged into the Windows Azure Portal and found the Hosted Services section.

We need a certificate in order to manage our cloud instance.  In this scenario, I am producing a certificate on my machine and sharing it with Windows Azure.  In a command prompt, I navigated to a directory where I wanted my physical certificate dropped.  I then executed the following command:

makecert -r -pe -a sha1 -n "CN=Windows Azure Authentication Certificate" -ss My -len 2048 -sp "Microsoft Enhanced RSA and AES Cryptographic Provider" -sy 24 testcert.cer

When this command completes, I have a certificate in my directory and see the certificate added to the “Current User” certificate store.

Next, while still in the Certificate Viewer, I exported this certificate (with the private key) out as a PFX.  This file will be used with the Azure instance that gets generated by StreamInsight Austin.  Back in the Windows Azure Portal, I navigated to the Management Certificates section and uploaded the CER file to the Azure subscription associated with StreamInsight Austin.

After this, I made sure that I had a “storage account” defined beneath my Windows Azure account.  This account is used by StreamInsight Austin and deployment fails if no such account exists.

Finally, I had to create a hosting service underneath my Azure subscription.  The window that pops up after clicking the New Hosted Service button on the ribbon lets you put a service under a subscription and define the deployment options and URL.  Note that I’ve chosen the “do not deploy” option since I have no package to upload to this instance.

The last pre-deployment step is to associate the PFX certificate with this newly created Azure instance.  When doing so, you must provide the password set when exporting the PFX file.

Next, I went to the Visual Studio solution provided with the StreamInsight Austin download.  There are a series of projects in this solution and the ones that I leveraged helped with provisioning the instance, deploying the StreamInsight application, and deleting the provisioned instance.  Note that there is a RESTful API for all of this and these Visual Studio projects just wrap up the operations into a C# API.

The provisioning project has a configuration file that must contain references to my specific Azure account.  These settings include the SubscriptionId (GUID associated with my Azure subscription), HostedServiceName (matching the Azure service I created earlier), StorageAccountName (name of the storage account for the subscription), StorageAccountKey (giant value visible by clicking “View Access Keys” on the ribbon), ServiceManagementCertificateFilePath (location on local machine where PFX file sits), ServiceManagementCertificatePassword (password provided for PFX file), and ClientCertificatePassword (value used when the provisioning project creates a new certificate).

Next, I ran the provisioning project which created a new certificate and invoked the StreamInsight Austin provisioning API that puts the StreamInsight binaries into an Azure instance.

When the provisioning is complete, you can see the newly created instance and certificates.

Neat.  It all completed in 5 or so minutes.  Also note that the newly created certificate is in the “My User” certificate store.

I then switched to the “deployment” project provided by StreamInsight Austin.  There are new components that get installed with StreamInsight Austin, including a Package class.  The Package contains references to all of the components that must be uploaded to the Windows Azure instance in order for the query to run.  In my case, I need the Azure AppFabric adapter, my “shared” component, and the Microsoft.ServiceBus.dll that the adapters use.

PKG.Package package = new PKG.Package("adapters");

package.AddResource(@"Seroter.AustinWalkthrough.SharedObjects.dll");
package.AddResource(@"Seroter.StreamInsight.AzureAppFabricAdapter.dll");
package.AddResource(@"Microsoft.ServiceBus.dll");

After updating the project to have the query from my previously built “onsite hosting” project, and updating the project’s configuration file to include the correct Azure instance URL and certificate password, I started up the deployment project.

You can see that my deployment is successful and my StreamInsight query was started.  I can use the RESTful APIs provided by StreamInsight Austin to check on the status of my provisioned instance.  By hitting a specific URL (https://azure.streaminsight.net/HostedServices/{serviceName}/Provisioning), I see the details.

With the query started, I turned on my Azure AppFabric listener service (who receives events from StreamInsight), and my service caller.  The data should flow to the Azure AppFabric endpoint, through StreamInsight Austin, and back out to an Azure AppFabric endpoint.

Content that everything works, and scared that I’d incur runaway hosting charges, I ran the “delete” project with removed my Azure instance and all traces of the application.

All in all, it’s a fairly straightforward effort.  Your onsite StreamInsight application transitions seamlessly to the cloud.  As mentioned in the first post of the series, the big caveat is that you need event sources that are accessible by the cloud instance.  I leveraged Windows Azure AppFabric to receive events, but you could also do a batch load from an internet-accessible database or file store.

When would you use StreamInsight Austin?  I can think of a few scenarios that make sense:

• First and foremost, if you have a wide range of event sources, including cloud hosted ones, having your complex event processing engine close to the data and easily accessible is compelling.
• Second, Austin makes good sense for variable workloads.  We can run the engine when we need to, and if it only operates on batch data, can shut it down when not in use.  This scenario will even more compelling once the transparent and elastic scale out of StreamInsight Austin is in place.
• Third, we can use it for proof-of-concept scenarios without requiring on-premises hardware.  By using a service instead of maintaining on-site hardware you offload your management and maintenance to StreamInsight Austin.

StreamInsight Austin is slated for a public CTP release later this year, so keep an eye out for more info.

StreamInsight is Microsoft’s (complex) event processing engine which takes in data and does in-memory pattern matching with the goal of uncovering real-time insight into information.  The StreamInsight team at Microsoft recently announced their upcoming  capability (code named “Austin”) to deploy StreamInsight applications to the Windows Azure cloud.  I got my hands on the early bits for Austin and thought I’d walk through an example of building, deploying and running a cloud-friendly StreamInsight application.  You can find the source code here.

You may recall that the StreamInsight architecture consists of input/output adapters and any number of “standing queries” that the data flows over.  In order for StreamInsight Austin to be effective, you need a way for the cloud instance to receive input data.  For instance, you could choose to poll a SQL Azure database or pull in a massive file from an Amazon S3 bucket.  The point is that the data needs to be internet accessible.  If you wish to push data into StreamInsight, then you must expose some sort of endpoint on the Azure instance running StreamInsight Austin.  Because we cannot directly host a WCF service on the StreamInsight Austin instance, our best bet is to use Windows Azure AppFabric to receive events.  In this post, I’ll show you how to build an Azure AppFabric adapter for StreamInsight.  In the next post, I’ll walk through the steps to deploy the on-premises StreamInsight application to Windows Azure and StreamInsight Austin.

As a reference point, the final solution looks like the picture below.  I have a client application which calls an Azure AppFabric Service Bus endpoint started up by StreamInsight Austin, and then take the output of StreamInsight query and send it through an adapter to an Azure AppFabric Service Bus endpoint that relays the message to a subscribing service.

I decided to use the product team’s WCF sample adapter as a foundation for my Azure AppFabric Service Bus adapter.  However, I did make a number of changes in order to simplify it a bit. I have one Visual Studio project that contains shared objects such as the input WCF contract, output WCF contract and StreamInsight Point Event structure.  The Point Event stores a timestamp and dictionary for all the payload values.

[DataContract]
    public struct WcfPointEvent
    {
        ///
 /// Gets the event payload in the form of key-value pairs. ///
        [DataMember]
        public Dictionary Payload { get; set; }

        ///
 /// Gets the start time for the event. ///
        [DataMember]
        public DateTimeOffset StartTime { get; set; }

        ///
 /// Gets a value indicating whether the event is an insert or a CTI. ///
        [DataMember]
        public bool IsInsert { get; set; }
    }

Each receiver of the StreamInsight event implements the following WCF interface contract.

[ServiceContract]
    public interface IPointEventReceiver
    {
        ///
 /// Attempts to dequeue a given point event. The result code indicates whether the operation /// has succeeded, the adapter is suspended -- in which case the operation should be retried later -- /// or whether the adapter has stopped and will no longer return events. ///
        [OperationContract]
        ResultCode PublishEvent(WcfPointEvent result);
    }

The service clients which send messages to StreamInsight via WCF must conform to this interface.

[ServiceContract]
    public interface IPointInputAdapter
    {
        ///
 /// Attempts to enqueue the given point event. The result code indicates whether the operation /// has succeeded, the adapter is suspended -- in which case the operation should be retried later -- /// or whether the adapter has stopped and can no longer accept events. ///
        [OperationContract]
        ResultCode EnqueueEvent(WcfPointEvent wcfPointEvent);
    }

I built a WCF service (which will be hosted through the Windows Azure AppFabric Service Bus) that implements the IPointEventReceiver interface and prints out one of the values from the dictionary payload.

public class ReceiveEventService : IPointEventReceiver
    {
        public ResultCode PublishEvent(WcfPointEvent result)
        {
            WcfPointEvent receivedEvent = result;
            Console.WriteLine("Event received: " + receivedEvent.Payload["City"].ToString());

            result = receivedEvent;
            return ResultCode.Success;
        }
    }

Now, let’s get into the StreamInsight Azure AppFabric adapter project.  I’ve defined a “configuration object” which holds values that are passed into the adapter at runtime.  These include the service address to host (or consume) and the password used to host the Azure AppFabric service.

public struct WcfAdapterConfig
    {
        public string ServiceAddress { get; set; }
        public string Username { get; set; }
        public string Password { get; set; }
    }

Both the input and output adapters have the required factory classes and the input adapter uses the declarative CTI model to advance the application time.  For the input adapter itself, the constructor is used to initialize adapter values including the cloud service endpoint.

public WcfPointInputAdapter(CepEventType eventType, WcfAdapterConfig configInfo)
{
this.eventType = eventType;
this.sync = new object();

// Initialize the service host. The host is opened and closed as the adapter is started
// and stopped.
this.host = new ServiceHost(this);
//define cloud binding
BasicHttpRelayBinding cloudBinding = new BasicHttpRelayBinding();
//turn off inbound security
cloudBinding.Security.RelayClientAuthenticationType = RelayClientAuthenticationType.None;

//add endpoint
ServiceEndpoint endpoint = host.AddServiceEndpoint((typeof(IPointInputAdapter)), cloudBinding, configInfo.ServiceAddress);
//define connection binding credentials
TransportClientEndpointBehavior cloudConnectBehavior = new TransportClientEndpointBehavior();
cloudConnectBehavior.CredentialType = TransportClientCredentialType.SharedSecret;
cloudConnectBehavior.Credentials.SharedSecret.IssuerName = configInfo.Username;
cloudConnectBehavior.Credentials.SharedSecret.IssuerSecret = configInfo.Password;
endpoint.Behaviors.Add(cloudConnectBehavior);

// Poll the adapter to determine when it is time to stop.
this.timer = new Timer(CheckStopping);
this.timer.Change(StopPollingPeriod, Timeout.Infinite);
}

On “Start()” of the adapter, I start up the WCF host (and connect to the cloud).  My Timer checks the state of the adapter and if the state is “Stopping”, the WCF host is closed.  When the “EnqueueEvent” operation is called by the service client, I create a StreamInsight point event and take all of the values in the payload dictionary and populate the typed class provided at runtime.

foreach (KeyValuePair keyAndValue in payload)
 {
       //populate values in runtime class with payload values
       int ordinal = this.eventType.Fields[keyAndValue.Key].Ordinal;
       pointEvent.SetField(ordinal, keyAndValue.Value);
  }
 pointEvent.StartTime = startTime;

 if (Enqueue(ref pointEvent) == EnqueueOperationResult.Full)
 {
        Ready();
 }

There is a fair amount of other code in there, but those are the main steps.  As for the output adapter, the constructor instantiates the WCF ChannelFactory for the IPointEventReceiver contract defined earlier.  The address passed in via the WcfAdapterConfig is applied to the Factory.  When StreamInsight invokes the Dequeue operation of the adapter, I pull out the values from the typed class and put them into the payload dictionary of the outbound message.

// Extract all field values to generate the payload.
result.Payload = this.eventType.Fields.Values.ToDictionary(
        f => f.Name,
        f => currentEvent.GetField(f.Ordinal));

//publish message to service
client = factory.CreateChannel();
client.PublishEvent(result);
((IClientChannel)client).Close();

I now have complete adapters to listen to the Azure AppFabric Service Bus and publish to endpoints hosted on the Azure AppFabric Service Bus.

I’ll now build an on-premises host to test that it all works.  If it does, then the solution can easily be transferred to StreamInsight Austin for cloud hosting.  I first defined the typed class that defines my event.

public class OrderEvent
    {
        public string City { get; set; }
        public string Product { get; set; }
    }

Recall that my adapter doesn’t know about this class.  The adapter works with the dictionary object and the typed class is passed into the adapter and translated at runtime.  Next up is setup for the StreamInsight host.  After creating a new embedded application, I set up the configuration object representing both the input WCF service and output WCF service.

//create reference to embedded server
using (Server server = Server.Create("RSEROTER"))
{

		//create WCF service config
     WcfAdapterConfig listenWcfConfig = new WcfAdapterConfig()
      {
          Username = "ISSUER",
          Password = "PASSWORD",
          ServiceAddress = "https://richardseroter.servicebus.windows.net/StreamInsight/RSEROTER/InputAdapter"
       };

     WcfAdapterConfig subscribeWcfConfig = new WcfAdapterConfig()
     {
           Username = string.Empty,
           Password = string.Empty,
           ServiceAddress = "https://richardseroter.servicebus.windows.net/SIServices/ReceiveEventService"
     };

     //create new application on the server
     var myApp = server.CreateApplication("DemoEvents");

     //get reference to input stream
     var inputStream = CepStream.Create("input", typeof(WcfInputAdapterFactory), listenWcfConfig, EventShape.Point);

     //first query
     var query1 = from i in inputStream
                            select i;

     var siQuery = query1.ToQuery(myApp, "SI Query", string.Empty, typeof(WcfOutputAdapterFactory), subscribeWcfConfig, EventShape.Point, StreamEventOrder.FullyOrdered);

     siQuery.Start();
    Console.WriteLine("Query started.");

    //wait for keystroke to end
    Console.ReadLine();

    siQuery.Stop();
    host.Close();
    Console.WriteLine("Query stopped. Press enter to exit application.");
    Console.ReadLine();

This is now a fully working, cloud-connected, onsite StreamInsight application.  I can take in events from any internal/external service caller and publish output events to any internal/external service.  I find this to be a fairly exciting prospect.  Imaging taking events from your internal Line of Business systems and your external SaaS systems and looking for patterns across those streams.

Looking for the source code?  Well here you go.  You can run this application today, whether you have StreamInsight Austin or not.  In the next post, I’ll show you how to take this application and deploy it to Windows Azure using StreamInsight Austin.