Chapter 8 Rx 扩展

Itminus.Tags.RxExtensions为ITag增加Rx支持。在开始之前，需要添加包引用：

dotnet add package Itminus.Tags.RxExtensions

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

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

观测从底层读取事件

本质上，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作为参数，可以有效避免这种长时间观测不到值的情况。