CODEWITHSUNDEEP
c#streaminsight
Matt Austin
Matt Austin

Reputation: 249

Stream Insight - Filter based on Type of Event

So I am fairly new to Stream Insight and this might be a fairly basic question but it seems that filtering queries by event type is not supported (see below) so how should this be done?

To flesh out my question a bit, this is my CepStream...

var stockStream = CepStream<MarketDataPoint>.Create(appName, typeof(MarketdataPointInputAdapterFactory), new MarketDataAdapterConfig(), EventShape.Point);

.. and this is my query ...

var patternResult = from w in stockStream.Where(x => x is FxSpot)
                                            .AlterEventDuration(e => TimeSpan.FromMinutes(1))
                                            .SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
                                        select new MarketDataPointEvent() {Price = w.Max(e => e.Price)};

...which yields this exception when run: The following expression is not supported by Microsoft Complex Event Processing LINQ Provider: (x is FxSpot)

As I understand it, I could create multiple adapters than effectively perform the filter but this doesn't seem like the way it "should" be done.

Any advice would be greatly appreciated!

Matt

Upvotes: 0

Views: 369

Answers (2)

TXPower275
TXPower275

Reputation: 511

The payload type you've defined is MarketDataPoint. This defines the schema for the event and according to Event Structure: "Events in the StreamInsight server are an ordered list of fields instead of .NET structs, which do not impose an order on its fields."

A potential workaround for what you are trying to accomplish is to include all the needed fields in MarketDataPoint that comprise an FxSpot. Then add an additional property to MarketDataPoint to define what type of MarketDataPoint it is.

Upvotes: 0

Donaciano Rojas
Donaciano Rojas

Reputation: 36

I would try to stay way from the is operator for type comparison. This is a sample using LINQPad, which I highly recommend to use. In this case the filter is done by stock. CTIs are automatically inserted by using

AdvanceTimeSettings.IncreasingStartTime

Sample

void Main()     
{

    var marketReadings = new []
                    {
                    new MarketDataPoint() { Stock = "MSFT", Value = 30.0, Timestamp = DateTime.Parse("11/19/2012 4:10:00 PM") },
                    new MarketDataPoint() { Stock = "MSFT", Value = 30.1, Timestamp = DateTime.Parse("11/19/2012 4:11:00 PM") },
                    new MarketDataPoint() { Stock = "GOOG", Value = 667.97, Timestamp = DateTime.Parse("11/19/2012 4:12:00 PM") },
                    new MarketDataPoint() { Stock = "GOOG", Value = 667.51, Timestamp = DateTime.Parse("11/19/2012 4:12:00 PM") },
                };

            var stockStream = marketReadings.ToPointStream(Application, e=> PointEvent.CreateInsert(e.Timestamp,e),AdvanceTimeSettings.IncreasingStartTime);  

            var patternResult = from w in stockStream.Where(x => x.Stock == "MSFT")
                                                    .AlterEventDuration(e => TimeSpan.FromMinutes(1))
                                                    .SnapshotWindow(SnapshotWindowOutputPolicy.Clip)
                                                select new  {Price = w.Max(e => e.Value)}   ;

            patternResult.Dump();   
        }


        public class MarketDataPoint
        {
            public string Stock { get; set; }
            public double Value { get; set; }
            public DateTime Timestamp { get;set;}

        }

}

Upvotes: 1

Related Questions