Rx operator to distinct sequences

IMPORTANT: for a description of the results and some more details, please have a look also to my answer

I need to group and filter a sequence of objects/events that usually are replicated, buffering them with a TimeSpan interval. I try to explain it better with sort of marble diagrams:

X-X-X-X-X-Y-Y-Y-Z-Z-Z-Z-X-X-Y-Z-Z

would produce

X---Y---Z---X---Y---Z

where X, Y and Z are different event types, and '---' means the interval. Additionally, I would also like to distinct by a key property that it is available on all types because they have a common base class:

X, Y, Z : A

and A contains a property Key. Using the notation X.a meaning X.Key = a, A final sample would be:

X.a-X.b-X.a-Y.b-Y.c-Z.a-Z.a-Z.c-Z.b-Z.c

would produce

X.a-X.b---Y.b-Y.c-Z.a-Z.c-Z.b

Can anybody help me putting together the required Linq operators (probably DistinctUntilChanged and Buffer) to achieve this behavior? Thanks

UPDATE 18.08.12:

as requested, I try to give a better explanation. We have devices collecting and sending events to a web service. These devices have an old logic (and we can't change it due to backward compatibility) and they continuously send an event until they receive an acknowledge; after the acknowledge, they send the next event in their queue, and so on. Events contain the network address of the unit and some other properties distinguishing events in the queue for each device. An event looks like this:

class Event
{
    public string NetworkAddress { get; }

    public string EventCode { get; }

    public string AdditionalAttribute { get; }
}

The goal is that of processing every 5 seconds the distinguished events received from all devices, storing information in the database (that's why we don't want to do it in ba

c# system.reactive

Edit Report

I'm not sure if this does exactly what you'd like, but you may be to group the elements explicitly using the group keyword, and then to manipulate the various IObservables separately before recombining them.

E.g. if we have class definitions such as

class A
{
    public char Key { get; set; }
}

class X : A { }
...

and a Subject<A>

Subject<A> subject = new Subject<A>();

then we can write

var buffered =
    from a in subject
    group a by new { Type = a.GetType(), Key = a.Key } into g
    from buffer in g.Buffer(TimeSpan.FromMilliseconds(300))
    where buffer.Any()
    select new
    {
        Count = buffer.Count,
        Type = buffer.First().GetType().Name,
        Key = buffer.First().Key
    };

buffered.Do(Console.WriteLine).Subscribe();

We can test this with the data you provided:

subject.OnNext(new X { Key = 'a' }); 
Thread.Sleep(100);
subject.OnNext(new X { Key = 'b' }); 
Thread.Sleep(100);
subject.OnNext(new X { Key = 'a' }); 
Thread.Sleep(100);
...
subject.OnCompleted();

To get the output you provided:

{ Count = 2, Type = X, Key = a }
{ Count = 1, Type = X, Key = b }
{ Count = 1, Type = Y, Key = b }
{ Count = 1, Type = Y, Key = c }
{ Count = 2, Type = Z, Key = a }
{ Count = 2, Type = Z, Key = c }
{ Count = 1, Type = Z, Key = b }

answered 2012-08-11T21:25:31.520

Rx operator to distinct sequences

1 Answer

Your Answer