Playing with mouse moves - and getting a serialization error

Question

We're looking into using StreamInsight in a sensor based application, but to start simple and learning the technology I wanted to use mouse moves as sample sensor readings. Copy & paste from code used in Rx projects and StreamInsight samples, I ended up with the following code:

// The mouse events from a WPF grid named "layoutRoot"
var mouseDown = from evt in Observable.FromEventPattern<MouseButtonEventArgs>(layoutRoot, "MouseLeftButtonDown") select evt.EventArgs.GetPosition(layoutRoot);
var mouseUp = from evt in Observable.FromEventPattern<MouseButtonEventArgs>(layoutRoot, "MouseLeftButtonUp") select evt.EventArgs.GetPosition(layoutRoot);
var mouseMove = from evt in Observable.FromEventPattern<MouseEventArgs>(layoutRoot, "MouseMove") select evt.EventArgs.GetPosition(layoutRoot);

// Left button down -> get the data
var res = from startLocation in mouseDown
            from endlocation in mouseMove.TakeUntil(mouseUp)
            select new
            {
                myTime = DateTime.Now,
                X = endlocation.X - startLocation.X,
                Y = endlocation.Y - startLocation.Y,
                D = Math.Sqrt(Math.Pow((endlocation.X - startLocation.X), 2) + Math.Pow((endlocation.Y - startLocation.Y), 2))
            };

// Observe on the dispather for putting data in a textbox on the GUI thread
res.ObserveOnDispatcher().Subscribe(val =>
{
    tbValue.Text = val.myTime.ToString("HH:mm:ss.ff") + ": " + val.X.ToString() + "; " + val.Y.ToString() + "; " + val.D.ToString("F1");
});

// StreamInsight
using (Server server = Server.Create("Instance1"))
{
    app = server.CreateApplication("app");

    // This should create a IQStreamable "stream" from the IObservable "res"
    var stream = app.DefineObservable(() => res).ToPointStreamable(
            r => PointEvent.CreateInsert(r.myTime, r),
            AdvanceTimeSettings.StrictlyIncreasingStartTime);

    // Doing some calculations
    double timeInterval = 1.0;
    var alteredForward = stream.AlterEventStartTime(s => s.StartTime.AddSeconds(timeInterval));

    var speeds = from evt in stream
                    from prev in alteredForward
                    select (evt.D - prev.D) / timeInterval;

    var alarms = from s in speeds where s > 10 select s;

    // This causes a serialization runtime error:
    // Cannot serialize value of type 'System.IObservable`1[<>f__AnonymousType0`4[System.DateTime,System.Double,System.Double,System.Double]]'
    foreach (var alarm in alarms.ToEnumerable())
    {
        Console.WriteLine(alarm);
    }
}

I've tried to find solutions, and I guess it's just trivial, but without any luck. Can somebody please help with an explanation and enlighten my understanding...

Pål

Answer 1

I'm not exactly sure what you are trying to prove out here. Your "test scenario" really bears no architectural resemblance to monitoring sensors - UI's, because of threading issues, always introduce some wacky threading issues when dealing with something as inherently and completely multi-threaded as StreamInsight. I would suggest that you use the Random Generator that you'll find in the StreamInsight samples on CodePlex instead.

That said, you should look at the inner exception that comes with the error:

Type 'System.Reactive.AnonymousObservable`1[<>f__AnonymousType0`2[System.DateTime,System.Windows.Point]]' cannot be serialized. Consider marking it with the DataContractAttribute attribute, and marking all of its members you want serialized with the DataMemberAttribute attribute.  If the type is a collection, consider marking it with the CollectionDataContractAttribute.  See the Microsoft .NET Framework documentation for other supported types.

Types are serialized into the StreamInsight engine - your native .NET types do not exist inside the engine. Instead they are serialized in and out of the engine, where they are represented in StreamInsight's own internal format (which isn't documented). This serialization is failing on your initial events. This piece:

            select new
            {
                myTime = DateTime.Now,
                X = endlocation.X - startLocation.X,
                Y = endlocation.Y - startLocation.Y,
                D = Math.Sqrt(Math.Pow((endlocation.X - startLocation.X), 2) + Math.Pow((endlocation.Y - startLocation.Y), 2))
            };

Also, it looks like you are trying to compare previous to current ... is that correct? If so, your query is wrong; you'd get all kinds of duplicated joins in the real world. You need to use the Alter/Clip/Shift (Fold Pairs) pattern. It would look something like the following:

var rateChange = from current in stream
                             from prev in stream
                                .AlterEventDuration(a => TimeSpan.MaxValue)
                                .ClipEventDuration(stream,(e1,e2) => e1.Id == e2.Id)
                                .ShiftEventTime(b => TimeSpan.FromTicks(1))
                             select new
                             {
                                 Current = current,
                                 Previous = prev
                             };

---

DevBiker (aka J Sawyer)
Microsoft MVP - Sql Server (StreamInsight)

---

---

If I answered your question, please mark as answer.
If my post was helpful, please mark as helpful.

Answer 2

Thanks for your reply! Using the mouse as a sensor was just for fun, and I see your comment about the threading issue. The real world sensor data we're analyzing are temperature and pressure readings, position indicators, etc. We're typically looking for patterns from multiple sensors - multi sensor data fusion, indicating e.g. machine condition. My hope is that StreamInsight could be used for a (close to) real time analysis.

It was your talk on .Net Rocks! that got me aware of StreamInsight from MS, and I will definitivly look more into the details in your answer.

Pål

---