BlogStreamInsightAustin03

gistfile1.cs

// Convert points to signals
var fooEdgeEvents = 
  from e in fooPointEvents
  .AlterEventDuration(e => TimeSpan.MaxValue)
  .ClipEventDuration(fooMessagesObservable, (e1, e2) => (e1.MessageId == e2.MessageId))
  select e;

var latencyQuery =
    from e1 in fooEdgeEvents
    join e2 in barPointEvents
    on e1.MessageId equals e2.MessageId
    select new
    {
        StoreId = e1.StoreId,
        LatencyMs = e2.Timestamp.Subtract(e1.Timestamp).Milliseconds
    };

var averageLatencyQuery = 
    from e1 in latencyQuery
    group e1 by e1.StoreId
    into storeGroup from win in storeGroup.TumblingWindow(TimeSpan.FromSeconds(60))
    select new MessageLatency()
    {
        Id = null,
        StoreId = storeGroup.Key,
        AverageLatency = win.Avg(e => e.LatencyMs),
        DateTime = DateTime.UtcNow
    };