Chapter 5 Rx 扩展

我们为ITag增加了Rx支持。在开始之前，根据你的需求，需要添加下面两个包的引用(之一)：

dotnet add package StdUnit.Tags.RxExtensions
dotnet add package StdUnit.Tags.R3Extensions

RxExtensions是使用dotnet/reactive来扩展ITag，而R3Extensions使用的则是Cysharp/R3。二者思路一致、接口类似，我们这里以dotnet/reactive为例讲解。

`ITag`的扩展

目前我们针对ITag主要提供了三个扩展方法：

tag.ObserveOnRead(): 观测从底层读取
tag.ObserveOnWritten(): 观测向底层写入
tag.Watch(): 观测测点发生了变化

这三个方法都是建立在ITag的测点读写事件上的。

观测从底层读取事件

本质上，ITag定义了一个事件OnTagRead：

public interface ITag 
{
    //...

    /// <summary>
    /// 在每次从底层读取后，触发事件
    /// </summary>
    event TagSyncEventHandler OnTagRead
}

ObserveOnRead()无非是把这个事件转成了Rx上可观测的事件流：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> ObserveOnRead(this ITag tag)
{
    var read = Observable.FromEventPattern<TagSyncEventHandler, ITag, TagSyncEventArgs>(
        h => tag.OnTagRead += h,
        h => tag.OnTagRead -= h
        );
    return read;
}

观测向底层写入事件

类似地，ITag也定义了一个事件OnTagWritten：

public interface ITag 
{
    //...

    /// <summary>
    /// 在每次向底层写入后，触发事件
    /// </summary>
    public event TagSyncEventHandler OnTagWritten;
}

ObserveOnWritten()也无非是把这个事件转成了Rx上可观测的事件流：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> ObserveOnWritten(this ITag tag)
{
    var written = Observable.FromEventPattern<TagSyncEventHandler, ITag, TagSyncEventArgs>(
        h => tag.OnTagWritten += h,
        h => tag.OnTagWritten -= h
        );
    return written;
}

观测测点改变事件

不管是从底层读取，还是向底层写入，都可以视作“内存”和“底层”之间发生了“同步”。tag.Watch()就是组合了这两种事件的一个便利方法：

public static IObservable<EventPattern<ITag, TagSyncEventArgs>> Watch(this ITag tag, bool startWithCurrent=true)
{
    var read = tag.ObserveOnRead();
    var written = tag.ObserveOnWritten();
    var obs = read.Merge(written);
    if(startWithCurrent)
    {
        var ev = new EventPattern<ITag, TagSyncEventArgs>(
            tag,
            new TagSyncEventArgs(tag.Value, tag.Timestamp, TagSyncEventArgs.Kinds.None)
            );
        return obs.StartWith(ev);
    }
    else
    {
        return obs;
    }
}

这里的默认值true表示需要把当时的值作为初始事件发给下游监听者。有些测点长时间不读写，而监听又通常又发生比较晚，指定true作为参数，可以有效避免这种长时间观测不到值的情况。

注意：不要在订阅中直接通过tag.Value=xx的方式修改点位。你应该通过外部意图的方式完成这种修改。关于这一点，我们会在后面详细讲解。

`ITagsProjectCtrl`的扩展

ITagsProjectCtrl有一个启停事件：

    event TagsProjectStartedOrStopped? StartedOrStopped;

上面这个委托TagsProjectStartedOrStopped的签名是：

public delegate void TagsProjectStartedOrStopped(ITagsProjectCtrl sender, TagsProjectEventArgs args);

这里的TagsProjectEventArgs提供了两个属性：

public class TagsProjectEventArgs : EventArgs
{
    /// 关联的项目
    public ITagsProject? Project { get; }
    /// true 表示启动事件， false表示停止事件
    public bool IsStarted { get; }
}

有了这个启停事件机制，我们就可以将它转成对事件的观测ctrl.ObserveStartedOrStopped():

public static Observable<TagsProjectEventArgs> ObserveStartedOrStopped(this ITagsProjectCtrl ctrl)
{
    var obs = Observable.FromEvent<TagsProjectStartedOrStopped, TagsProjectEventArgs>(
        h => (sender, e) => h(e),
        d => ctrl.StartedOrStopped += d,
        d => ctrl.StartedOrStopped -= d
    );
    return obs;
}

StdUnit.Tags