Как использовать Reactive Extensions для кэширования, регулирования и ретрансляции нескольких событий?
Я пытаюсь изучить новую платформу Reactive Extensions для.Net, и это звучит как идеальное решение для моего приложения. Изучив примеры (и все еще довольно слабые с LINQ), я изо всех сил пытаюсь выяснить, как использовать инфраструктуру RX для решения задачи ниже.
Цель состоит в том, чтобы создать настраиваемое событие "ретрансляция" между пользовательским источником данных и графическим интерфейсом. Реле будет использовать LINQ для тестирования и фильтрации входящих событий, кэшировать квалифицированные события в списке в ожидании следующего интервала времени, затем синхронизировать с потоком графического интерфейса и воспроизводить события в порядке их получения.
Как вы можете использовать RX для координации кэширования, фильтрации и ретрансляции нескольких событий, таких как события для добавления, изменения и удаления значений из пользовательского источника данных?
Это может требовать больших затрат, но любые рекомендации по подходу к этой проблеме будут очень благодарны. Смотрите пример кода ниже...
public delegate void EventDelegateAdd(Thing thing);
public delegate void EventDelegateChange(Thing thing);
public delegate void EventDelegateRemove(Thing thing);
public delegate void EventDelegateBulkChangesStart();
public delegate void EventDelegateBulkChangesEnd();
// The "Things" that are stored in MyCustomDataSource
public class Thing
{
public string Key { get; set; }
public string Title { get; set; }
public object OtherStuff { get; set; }
}
// A custom observable data source with events that indicate when Things are
// added, changed, or removed.
public class MyCustomDataSource
{
public event EventDelegateAdd AddingThing;
public event EventDelegateChange ChangingThing;
public event EventDelegateRemove RemovingThing;
// The rest of the class that manages the database of Things...
}
// This class forms a configurable event bridge between the MyCustomDataSource and
// the GUI. The goal is to cache, filter, and throttle the events so that the GUI
// updates only occasionally with bulk changes that are relevant for that control.
public class MyEventCachingBridge
{
private MyCustomDataSource mSource;
public event EventDelegateAdd AddingThing;
public event EventDelegateChange ChangingThing;
public event EventDelegateRemove RemovingThing;
public event EventDelegateBulkChangesStart BulkChangesStart;
public event EventDelegateBulkChangesEnd BulkChangesEnd;
public MyEventCachingBridge(MyCustomDataSource source, int eventRelayInterval)
{
mSource = source;
// Magical Reactive Extensions code goes here that subscribes to all 3 events...
//
// mSource.AddingThing
// mSource.ChangingThing
// mSource.RemovingThing
//
// ...filters and records a list of the events as they are received ( maintaining order of events too ),
// then every eventRelayInterval milliseconds, plays back the events in bulk to update the GUI
// ( on the GUIs thread ). Note that LINQ will be used to filter the Things so that a subset of
// Thing changes are relayed to the GUI - i.e. - not all Thing events are observed by the GUI.
}
public void PlayBackCachedEvents()
{
BulkChangesStart(); // Raise Event to notify GUI to suspend screen updates
// Play back the list of events to push changes to ListView, TreeView, graphs, etc...
//
// this.AddingThing(Thing); // Fire events in order received
// this.ChangingThing(Thing); // Fire events in order received
// this.RemovingThing(Thing); // Fire events in order received
BulkChangesEnd(); // Raise Event to notify GUI to allow control refresh
}
Учитывая описанную задачу, я не уверен, что должно быть скоординировано в общем коде класса, а что должно быть скрыто в утверждениях RX.
Я также ценю тот факт, что объединение трех событий в одно событие с перечислением, указывающим цель события, значительно упростит ситуацию, однако в реальном приложении можно кэшировать гораздо больше, чем три события. Я не хочу, чтобы накладные расходы приходилось использовать постоянное тестирование оператора Switch для идентификации каждого события. Там будут большие объемы событий, направленных на потенциально многочисленные интерфейсы GUI.
Спасибо за любые предложения.
2 ответа
Решение
Ах, я люблю играть с Rx... вот один подход; сначала по частям, потом по всему
РЕДАКТИРОВАТЬ: изменение в соответствии с комментариями
Во-первых, вам нужно настроить потоки для событий. Пока мы это делаем, давайте заменим "старый" шаблон событий.NET, который предоставляет мост, на "новый hawtness" Rx:
public Subject<Thing> BufferedAdds {get; private set;}
public Subject<Thing> BufferedChanges {get; private set;}
public Subject<Thing> BufferedRemoves {get; private set;}
_adds = Observable.FromEvent<EventDelegateAdd, Thing>(
ev => new EventDelegateAdd(ev),
h => mSource.AddingThing += h,
h => mSource.AddingThing -= h);
BufferedAdds = new Subject<Thing>();
_changes = Observable.FromEvent<EventDelegateChange, Thing>(
ev => new EventDelegateChange(ev),
h => mSource.ChangingThing += h,
h => mSource.ChangingThing -= h);
BufferedChanges = new Subject<Thing>();
_removes = Observable.FromEvent<EventDelegateRemove, Thing>(
ev => new EventDelegateRemove(ev),
h => mSource.RemovingThing += h,
h => mSource.RemovingThing -= h);
BufferedRemoves = new Subject<Thing>();
Мы также собираемся связать все с IScheduler сейчас - IScheduler использование является ключевым аспектом для тестирования этого кровавого материала без Thread.Sleep в повсюду - я настоятельно рекомендую исследовать эту тему!
public MyEventCachingBridge(
MyCustomDataSource source,
int eventRelayInterval,
IScheduler scheduler)
{
Затем вы захотите объединить все входящие события в один поток, а затем "разделить" этот поток на основе временного окна - Buffer Оператор идеально подходит для этого:
_buffer = Observable.Merge(scheduler,
_adds.Select(e => Tuple.Create(e, ThingEventType.Add)),
_changes.Select(e => Tuple.Create(e, ThingEventType.Change)),
_removes.Select(e => Tuple.Create(e, ThingEventType.Remove)))
.Buffer(TimeSpan.FromMilliseconds(eventRelayInterval), scheduler);
Обратите внимание, что я упаковываю тип события обратно в поток - это так, чтобы мы могли предпринять соответствующие действия во время воспроизведения - перечисление:
private enum ThingEventType
{
Add,
Change,
Remove
}
Так что теперь нам нужно что-то прослушивать и хранить пакетные события - здесь есть несколько вариантов, но давайте воспользуемся простым List с синхронизацией:
private Queue<IList<Tuple<Thing,ThingEventType>>> _eventQueue;
private static object SyncRoot = new object();
_eventQueue = new Queue<IList<Tuple<Thing,ThingEventType>>>();
// A serial disposable is a sort of "Disposable holder" - when you change it's
// Disposable member, it auto-disposes what you originally had there...no real
// need for it here, but potentially useful later
_watcherDisposable = new SerialDisposable();
_watcherDisposable.Disposable = _buffer
.ObserveOn(_scheduler)
.Subscribe(batch =>
{
lock(SyncRoot) { _eventQueue.Enqueue(batch); }
});
_disposables.Add(_watcherDisposable);
Давайте также "автоматически подключим" импульс воспроизведения к каждому eventRelayInterval Миз:
var pulse = Observable.Interval(
TimeSpan.FromMilliseconds(eventRelayInterval),
_scheduler);
_disposables.Add(pulse
.ObserveOn(_scheduler)
.Subscribe(x => PlayBackCachedEvents()));
Подписки всегда IDisposable, и вы захотите избавиться от них, поэтому давайте добавим некоторые вещи для этого:
public class MyEventCachingBridge : IDisposable
{
CompositeDisposable _disposables;
public void Dispose()
{
_disposables.Dispose();
}
А теперь для воспроизведения:
public void PlayBackCachedEvents()
{
BulkChangesStart(); // Raise Event to notify GUI to suspend screen updates
// Play back the list of events to push changes
lock(SyncRoot)
{
foreach(var batch in _eventQueue)
{
// Play back the list of events to push changes to ListView, TreeView, graphs, etc...
foreach(var evt in batch)
{
switch(evt.Item2)
{
case ThingEventType.Add: BufferedAdds.OnNext(evt.Item1); break;
case ThingEventType.Change: BufferedChanges.OnNext(evt.Item1);break;
case ThingEventType.Remove: BufferedRemoves.OnNext(evt.Item1);break;
}
}
}
_eventQueue.Clear();
}
BulkChangesEnd(); // Raise Event to notify GUI to allow control refresh
}
СЕЙЧАС - мы хотим, чтобы все было на стороне потребителя, поэтому давайте смоделируем окно пользовательского интерфейса (это WPF, отрегулируйте соответственно):
public class BridgeConsumer : Window, IDisposable
{
private readonly CompositeDisposable _disposables;
private IScheduler _scheduler;
private StackPanel _panel;
public void OnLoaded(object sender, RoutedEventArgs ea)
{
_panel = new StackPanel();
this.Content = _panel;
}
public BridgeConsumer(MyEventCachingBridge bridge, IScheduler scheduler)
{
// for cleanup of any subscriptions
_disposables = new CompositeDisposable();
_disposables.Add(bridge);
_scheduler = scheduler;
Loaded += OnLoaded;
// setup a listener for the bulk start/end events on the bridge
var bulkStart = Observable.FromEvent(
h => bridge.BulkChangesStart += new EventDelegateBulkChangesStart(h),
h => bridge.BulkChangesStart -= new EventDelegateBulkChangesStart(h));
var bulkEnd = Observable.FromEvent(
h => bridge.BulkChangesEnd += new EventDelegateBulkChangesEnd(h),
h => bridge.BulkChangesEnd -= new EventDelegateBulkChangesEnd(h));
// the "meaty bit" -
// 1. create a "window" defined by bulk start/end events
// 2. inside that "window", trap any occurrences on a
// merged view of adds/changes/removes
// 3. foreach event in that window, select that event
// (i.e., give us window contents as a stream of sorts)
var bridgeWatcher =
from thingEventWindow in
Observable.Merge(
bridge.BufferedAdds.Select(t => Tuple.Create("add", t)),
bridge.BufferedChanges.Select(t => Tuple.Create("change", t)),
bridge.BufferedRemoves.Select(t => Tuple.Create("remove", t))
)
.Window(bulkStart, start => bulkEnd)
from thingEvent in thingEventWindow
select thingEvent;
// this could just as easily be a method, a bound call to the viewmodel, etc
Action<Thing, string, DateTimeOffset> addToList = (thing, msg, ts) =>
{
var text = new TextBlock()
{
Text = string.Format(
"At:{0} Key:{2} Msg:{3} - nowTime = {1}",
thing.TimeStamp,
ts,
thing.Key,
msg) };
_panel.Children.Add(text);
};
_disposables.Add(bridgeWatcher
// CAREFUL! "ObserveOn" means what you'd think "SubscribeOn" would
.ObserveOnDispatcher()
.Subscribe(tup =>
{
addToList(tup.Item2, tup.Item1, _scheduler.Now);
}));
}
public void Dispose()
{
// clean up
if(_disposables != null) _disposables.Dispose();
}
}
Весь Шебанг:
void Main()
{
var scheduler = Scheduler.Default;
var rnd = new Random();
var canceller = new CancellationTokenSource();
var source = new MyCustomDataSource();
var eventRelayInterval = 2000;
var bridge = new MyEventCachingBridge(source, eventRelayInterval, scheduler);
var window = new BridgeConsumer(bridge);
window.Closed += (o,e) => { canceller.Cancel(); window.Dispose(); };
window.Show();
Task.Factory.StartNew(
() =>
{
while(true)
{
var thing = new Thing()
{
Key = "added thing " + rnd.Next(0, 100),
Title = "title for added thing",
TimeStamp = scheduler.Now.DateTime
};
source.FireAdd(thing);
Thread.Sleep(rnd.Next(1,10) * 100);
}
}, canceller.Token);
}
public class BridgeConsumer : Window, IDisposable
{
private readonly CompositeDisposable _disposables;
private StackPanel _panel;
public void OnLoaded(object sender, RoutedEventArgs ea)
{
_panel = new StackPanel();
this.Content = _panel;
}
public BridgeConsumer(MyEventCachingBridge bridge)
{
_disposables = new CompositeDisposable();
_disposables.Add(bridge);
Loaded += OnLoaded;
var bulkStart = Observable.FromEvent(
h => bridge.BulkChangesStart += new EventDelegateBulkChangesStart(h),
h => bridge.BulkChangesStart -= new EventDelegateBulkChangesStart(h));
var bulkEnd = Observable.FromEvent(
h => bridge.BulkChangesEnd += new EventDelegateBulkChangesEnd(h),
h => bridge.BulkChangesEnd -= new EventDelegateBulkChangesEnd(h));
var bridgeWatcher =
from thingEventWindow in
Observable.Merge(
bridge.BufferedAdds.Select(t => Tuple.Create("add", t)),
bridge.BufferedChanges.Select(t => Tuple.Create("change", t)),
bridge.BufferedRemoves.Select(t => Tuple.Create("remove", t))
)
.Window(bulkStart, start => bulkEnd)
from thingEvent in thingEventWindow
select thingEvent;
Action<Thing, string> addToList = (thing, msg) =>
{
var text = new TextBlock()
{
Text = string.Format(
"At:{0} Key:{1} Msg:{2}",
thing.TimeStamp,
thing.Key,
msg)
};
_panel.Children.Add(text);
};
_disposables.Add(bridgeWatcher.ObserveOnDispatcher().Subscribe(tup =>
{
addToList(tup.Item2, tup.Item1);
}));
}
public void Dispose()
{
if(_disposables != null) _disposables.Dispose();
}
}
public delegate void EventDelegateAdd(Thing thing);
public delegate void EventDelegateChange(Thing thing);
public delegate void EventDelegateRemove(Thing thing);
public delegate void EventDelegateBulkChangesStart();
public delegate void EventDelegateBulkChangesEnd();
// The "Things" that are stored in MyCustomDataSource
public class Thing
{
public DateTime TimeStamp {get; set;}
public string Key { get; set; }
public string Title { get; set; }
public object OtherStuff { get; set; }
public override string ToString()
{
return string.Format("At:{0} Key:{1} Title:{2}", this.TimeStamp, this.Key, this.Title);
}
}
// A custom observable data source with events that indicate when Things are
// added, changed, or removed.
public class MyCustomDataSource
{
public event EventDelegateAdd AddingThing = delegate { };
public event EventDelegateChange ChangingThing = delegate { };
public event EventDelegateRemove RemovingThing = delegate { };
// The rest of the class that manages the database of Things...
public void FireAdd(Thing toAdd)
{
AddingThing(toAdd);
}
public void FireChange(Thing toChange)
{
ChangingThing(toChange);
}
public void FireRemove(Thing toRemove)
{
RemovingThing(toRemove);
}
}
// This class forms a configurable event bridge between the MyCustomDataSource and
// the GUI. The goal is to cache, filter, and throttle the events so that the GUI
// updates only occasionally with bulk changes that are relevant for that control.
public class MyEventCachingBridge : IDisposable
{
private enum ThingEventType
{
Add,
Change,
Remove
}
private MyCustomDataSource mSource;
private IScheduler _scheduler;
public event EventDelegateBulkChangesStart BulkChangesStart = delegate { };
public event EventDelegateBulkChangesEnd BulkChangesEnd = delegate { };
public IObservable<Thing> RawAdds {get; private set;}
public IObservable<Thing> RawChanges {get; private set;}
public IObservable<Thing> RawRemoves {get; private set;}
public Subject<Thing> BufferedAdds {get; private set;}
public Subject<Thing> BufferedChanges {get; private set;}
public Subject<Thing> BufferedRemoves {get; private set;}
private readonly IObservable<IList<Tuple<Thing,ThingEventType>>> _buffer;
private List<IList<Tuple<Thing,ThingEventType>>> _eventQueue;
private static object SyncRoot = new object();
private readonly CompositeDisposable _disposables;
private readonly SerialDisposable _watcherDisposable;
public MyEventCachingBridge(MyCustomDataSource source, int eventRelayInterval, IScheduler scheduler)
{
_disposables = new CompositeDisposable();
mSource = source;
_scheduler = scheduler;
_eventQueue = new List<IList<Tuple<Thing,ThingEventType>>>();
// Magical Reactive Extensions code goes here that subscribes to all 3 events...
//
// mSource.AddingThing
// mSource.ChangingThing
// mSource.RemovingThing
//
// ...filters and records a list of the events as they are received ( maintaining order of events too ),
// then every eventRelayInterval milliseconds, plays back the events in bulk to update the GUI
// ( on the GUIs thread ). Note that LINQ will be used to filter the Things so that a subset of
// Thing changes are relayed to the GUI - i.e. - not all Thing events are observed by the GUI.
RawAdds = Observable.FromEvent<EventDelegateAdd, Thing>(
ev => new EventDelegateAdd(ev),
h => mSource.AddingThing += h,
h => mSource.AddingThing -= h);
BufferedAdds = new Subject<Thing>();
RawChanges = Observable.FromEvent<EventDelegateChange, Thing>(
ev => new EventDelegateChange(ev),
h => mSource.ChangingThing += h,
h => mSource.ChangingThing -= h);
BufferedChanges = new Subject<Thing>();
RawRemoves = Observable.FromEvent<EventDelegateRemove, Thing>(
ev => new EventDelegateRemove(ev),
h => mSource.RemovingThing += h,
h => mSource.RemovingThing -= h);
BufferedRemoves = new Subject<Thing>();
_buffer = Observable.Merge(
_scheduler,
RawAdds.Select(e => Tuple.Create(e, ThingEventType.Add)),
RawChanges.Select(e => Tuple.Create(e, ThingEventType.Change)),
RawRemoves.Select(e => Tuple.Create(e, ThingEventType.Remove)))
.Buffer(TimeSpan.FromMilliseconds(eventRelayInterval), _scheduler);
_watcherDisposable = new SerialDisposable();
_watcherDisposable.Disposable = _buffer
.ObserveOn(_scheduler)
.Subscribe(batch =>
{
lock(SyncRoot) { _eventQueue.Add(batch); }
});
_disposables.Add(_watcherDisposable);
var pulse = Observable.Interval(TimeSpan.FromMilliseconds(eventRelayInterval), _scheduler);
_disposables.Add(pulse.ObserveOn(_scheduler).Subscribe(x => PlayBackCachedEvents()));
}
private void PlayBackCachedEvents()
{
BulkChangesStart(); // Raise Event to notify GUI to suspend screen updates
try
{
//_eventQueue.Dump();
lock(SyncRoot)
{
foreach(var batch in _eventQueue)
{
// Play back the list of events to push changes to ListView, TreeView, graphs, etc...
foreach(var evt in batch)
{
switch(evt.Item2)
{
case ThingEventType.Add: BufferedAdds.OnNext(evt.Item1); break;
case ThingEventType.Change: BufferedChanges.OnNext(evt.Item1);break;
case ThingEventType.Remove: BufferedRemoves.OnNext(evt.Item1);break;
}
}
}
_eventQueue.Clear();
}
}
catch(Exception ex)
{
Console.WriteLine("Exception during playback:" + ex);
}
BulkChangesEnd(); // Raise Event to notify GUI to allow control refresh
}
public void Dispose()
{
_disposables.Dispose();
}
}
Я немного опоздал на это конкретное шоу, но думаю, что вам нужен наблюдаемый кеш / список, который даст набор изменений для всех добавлений, обновлений и удалений.
Я реализовал это и сделал его открытым исходным кодом: