First StreamInsight attempt using Performance Counter

As my first attempt with StreamInsight, I decided to use the Performance Counter as my stream of data. Here is a brief overview of implementing the StreamInsight portion

3 important downloads to get started, with StreamInsight:

Download StreamInsight: http://www.microsoft.com/en-us/download/details.aspx?id=26720

Download Reactive: http://www.microsoft.com/en-us/download/confirmation.aspx?id=28568
Download Examples: http://streaminsight.codeplex.com/releases/view/90143


First I created an class to represent the data I wanted to capture in my stream. Not all data types can be used with StreamInsight (ie: enum). Here is a list of supported data types: http://msdn.microsoft.com/en-us/library/ee378905.aspx

Code Snippet

205. // Input events of CounterSample of supported data types
206. public class StreamableCounterSample
207. {
208.     public float CpuUtilization { get; set; }
209.     public long TimeStamp { get; set; }
210. }

The main portion of the program.
Here starting on line 86, I create my StreamInsight server embedded in memory. The server name is the instance name that you give after installing StreamInsight. With StreamInsight 2.1, they added IQStreamable which allows you to query the stream. Lines 104 & 105 is when the stream will be queried for the results.

Code Snippet

81. private float CalculatePerformance(BackgroundWorker worker, DoWorkEventArgs e)
82. {
83.     CreateCounters();
84.  
85.     //embedded (in-memory)
86.     using (Server server = Server.Create("StreamInsight21"))
87.     {
88.         Microsoft.ComplexEventProcessing.Application application = server.CreateApplication("app");
89.  
90.         //A query for reading events from a stream
91.         IQStreamable<StreamableCounterSample> inputStream = null;
92.  
93.         inputStream = CreateStream(application);
94.  
95.         while (true)
96.         {
97.             if (worker.CancellationPending)
98.             {
99.                 e.Cancel = true;
100.                 break;
101.             }
102.             else
103.             {
104.                 perf = inputStream.ToObservable().ToEnumerable().Last().CpuUtilization;
105.                 timestamp = DateTime.FromFileTime(inputStream.ToObservable().ToEnumerable().Last().TimeStamp);
106.                 worker.ReportProgress((int)perf);
107.             }
108.         }
109.  
110.         return inputStream.ToObservable().ToEnumerable().Last().CpuUtilization;
111.  
112.     }
113. }

When creating the stream in line 93, I'm setting up my observation. In this case CollectSamples is my source which is then converted to a temporal stream via ToPointStreamable, which inserts a single event instance with a datetime.

Code Snippet

150. static IQStreamable<StreamableCounterSample> CreateStream(Microsoft.ComplexEventProcessing.Application application)
151. {
152.     // Live data uses IQbservable<>
153.     return
154.         application.DefineObservable(() => CollectSamples()).ToPointStreamable(
155.         r => PointEvent<StreamableCounterSample>.CreateInsert(DateTime.Now, r),
156.         AdvanceTimeSettings.StrictlyIncreasingStartTime);
157. }

The collect samples, is where my data will be pulled. In this case I'm pulling my data from the performance counter, but returning an observable interval representing this temporal event that occurs every 1 second.

Code Snippet

137. private static IObservable<StreamableCounterSample> CollectSamples()
138. {
139.     List<StreamableCounterSample> data = new List<StreamableCounterSample>();
140.     
141.     data.Add(new StreamableCounterSample
142.     {
143.         CpuUtilization = perfCounter.NextValue(),
144.         TimeStamp = DateTime.Now.ToFileTime()
145.     });
146.  
147.     return ToObservableInterval(data, TimeSpan.FromMilliseconds(1000), Scheduler.ThreadPool);
148. }

Code Snippet

121. private static IObservable ToObservableInterval(IEnumerable source, TimeSpan period, IScheduler scheduler)
122. {
123.     return Observable.Using(
124.         () => source.GetEnumerator(),
125.         it => Observable.Generate(
126.             default(object),
127.             _ => it.MoveNext(),
128.             _ => _,
129.             _ =>
130.             {
131.                 //Console.WriteLine("Input {0}", it.Current);
132.                 return it.Current;
133.             },
134.             _ => period, scheduler));
135. }

In the end, the final product is a simple winform showing the current CPU Utilization and the last time it ran.

The program can be downloaded from: