【发布时间】：2019-12-27 12:53:44
【问题描述】：
新：带有测试的整个源代码现在位于https://github.com/bboyle1234/ReactiveTest
假设我们有一个视图状态对象，它可以通过小的部分视图更改事件进行更新。以下是总视图、增量视图更新事件和构建总视图的累加器函数Update 的一些示例模型：
interface IDeviceView : ICloneable {
Guid DeviceId { get; }
}
class DeviceTotalView : IDeviceView {
public Guid DeviceId { get; set; }
public int Voltage { get; set; }
public int Currents { get; set; }
public object Clone() => this.MemberwiseClone();
}
class DeviceVoltagesUpdateView : IDeviceView {
public Guid DeviceId { get; set; }
public int Voltage { get; set; }
public object Clone() => this.MemberwiseClone();
}
class DeviceCurrentsUpdateView : IDeviceView {
public Guid DeviceId { get; set; }
public int Current { get; set; }
public object Clone() => this.MemberwiseClone();
}
class DeviceUpdateEvent {
public DeviceTotalView View;
public IDeviceView LastUpdate;
}
static DeviceUpdateEvent Update(DeviceUpdateEvent previousUpdate, IDeviceView update) {
if (update.DeviceId != previousUpdate.View.DeviceId) throw new InvalidOperationException("Device ids do not match (numskull exception).");
var view = (DeviceTotalView)previousUpdate.View.Clone();
switch (update) {
case DeviceVoltagesUpdateView x: {
view.Voltage = x.Voltage;
break;
}
case DeviceCurrentsUpdateView x: {
view.Currents = x.Current;
break;
}
}
return new DeviceUpdateEvent { View = view, LastUpdate = update };
}
接下来，假设我们已经有一个可注入服务，它能够为所有设备生成可观察到的小更新事件流，并且我们想要创建一个可以为单个设备生成聚合视图流的服务。
这是我们要创建的服务的接口：
interface IDeviceService {
/// <summary>
/// Gets an observable that produces aggregated update events for the device with the given deviceId.
/// On subscription, the most recent event is immediately pushed to the subscriber.
/// There can be multiple subscribers.
/// </summary>
IObservable<DeviceUpdateEvent> GetDeviceStream(Guid deviceId);
}
如何使用System.Reactive v4 库中的响应式扩展来实现此接口及其要求，以.netstandard2.0 为目标？这是我与 cmets 的锅炉代码，这是我所能得到的。
class DeviceService : IDeviceService {
readonly IObservable<IDeviceView> Source;
public DeviceService(IObservable<IDeviceView> source) { // injected parameter
/// When injected here, "source" is cold in the sense that it won't produce events until the first time it is subscribed.
/// "source" will throw an exception if its "Subscribe" method is called more than once as it is intended to have only one observer and
/// be read all the way from the beginning.
Source = source;
/// Callers of the "Subscribe" method below will expect data to be preloaded and will expect to be immediately delivered the most
/// recent event. So we need to immediately subscribe to "source" and start preloading the aggregate streams.
/// I'm assuming there is going to need to be a groupby to split the stream by device id.
var groups = source.GroupBy(x => x.DeviceId);
/// Now somehow we need to perform the aggregrate function on each grouping.
/// And create an observable that immediately delivers the most recent aggregated event when "Subscribe" is called below.
}
public IObservable<DeviceUpdateEvent> GetDeviceStream(Guid deviceId) {
/// How do we implement this? The observable that we return must be pre-loaded with the latest update
throw new NotImplementedException();
}
}
【问题讨论】：
标签： c# system.reactive