Используйте RxExtensions для агрегирования потока данных ⇐ C#

[Anonymous]

У меня есть игровой проект по изучению Rx, поскольку я всегда хотел это изучить и мог бы оказаться полезным.
Кажется, я просто не могу понять это так, как я понял Linq.
Поэтому я у вас есть поток данных, и вы хотели бы выполнить некоторую агрегацию потока.
По сути, сгруппируйте источник по имени, а затем сгруппируйте дополнительные наблюдаемые данные в окнах по 3 секунды, используя первый элемент в группе в качестве контрольной точки.
Итак, у меня есть накопленное окно данных за 3 секунды. Данные сгруппированы по имени.
Затем каждая группа группируется по людям с одинаковым количеством денег.
Эта подгруппа будет агрегирована или сокращена, чтобы получить один объект PersonData с средний возраст и сумма в долларах.
Кажется, все просто? но я просто не могу этого понять.

Код: Выделить всё

//PersonData to represent object event
class PersonData
{
public int Age;
public string Name;
public DateTime EntryTime;
public double DollarAmount;
}

Объект-наблюдатель для отслеживания сокращения

Код: Выделить всё

public class Observer : IObserver where T: PersonData
{
private T _t;
private int _count = 0;
private double _dollarAmt;

public void OnNext(T value)
{
_t = value; // just replace with last
_dollarAmt += value.DollarAmount;
++_count;
}

public void OnError(Exception error)
{
}

public void OnCompleted()
{
_dollarAmt /= _count;
}
}

class RxObservableSample
{
List
 _personSource = new List();

public RxObservableSample()
{
//group by name
// with a discrete window of 3 seconds
// this is not accurate, I would prefer to use 3 seconds window using the EntryTime as reference point
//from the grouped observable, aggregate the dollar amount of all PersonData.

//simple attempt
var windowedObservable = _personSource.ToObservable().GroupByUntil(x=>x.Name, x.Buffer(TimeSpan.FromSeconds(1)));

}

}

После windowedObservable я не знаю, куда идти дальше. Я не могу получить доступ к отдельному объекту данных.
Я пробовал агрегатировать, но он запрашивает начальное значение, но начальное значение должно быть первыми данными в 3-секундном окне, к которым я все равно не могу получить доступ.
Я пробовал заменить Buffer на Window, но это не имеет значения, поскольку я не могу получить доступ к каждому PersonData.
Есть идеи, как понять Rx? Есть подсказки, как действовать?

Подробнее здесь: https://stackoverflow.com/questions/788 ... am-of-data