c# Reactive Extension中的FromEventPattern和FromEvent

一些废话

在做一个设备的报告同步功能，通过FileSystemWatcher监控设备在指定位置生成的报告，解析并上传。
之所以监控文件系统，而不是跟设备软件直接对接，又会有一大段牢骚，暂且不提。

而监控文件时，文件的修改创建等事件，有时会被多次上报，这可能是文件系统或者代码的实现相关，所以要做防抖动。

之前通过自定义的私有字段/定时器做过防抖动，有些麻烦，而且不是很容易维护，在网上了解到Reactive Extension(对应.net core中的System.Reactive包），可以做防抖，于是找相关文档读了一下。

一开始看的一头雾水，主要是功能太多了，主要目的也不是防抖的，所以很不容易找到想要的内容（简单直接的例子）

正文

Observable从事件中创建，这里有两个创建方式FromEventPattern和FromEvent（其实还有很多，但是我涉及的是这两个），找了半天相关资料（SO上的这个回答帮助很大），并且写了一些测试代码，总算是明白大概咋回事了。

最终代码

先贴上最终可以使用的代码方法（其实正确方式不止一种）

var watcher = new FileSystemWatcher();
watcher.Path = "d:\test";

var observable = Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
            ev => watcher.Changed += ev,
            ev => watcher.Changed -= ev);

var d = observable.Throttle(TimeSpan.FromSeconds(5))
    .Subscribe(
        a => Console.WriteLine($"Watcher type:{a.EventArgs.ChangeType }, file:{a.EventArgs.Name }, sender: {a.Sender}"),
        ex => Console.WriteLine("observer error:" + ex.ToString()),
        () => Console.WriteLine("observer complete.")

Observable的实现理解

Observable用于绑定Event源和Observer中的delegate，他的作用在于可以进行一些过滤处理（个人理解）。

其中Observer主要通过实现OnNext这一个delegate，当Observable有事件触发时，就会调用Observer注册的OnNext方法，OnNext方法是Action<T>类型的，而模板参数T来自Observable创建时指定。

.net 的事件

.net中的事件处理函数符合EventHandler这个delegate格式

public delegate void EventHandler(object sender, EventArgs e);

传入两个参数，返回为空。

Observable用于将事件处理分发给observer，虽然事件可以是任何自定义事件，但是对于符合.net标准的EventHandler模式事件，提供了FromEventPattern方式创建Observable，而其他自定义模式的事件，需要使用FromEvent进行创建。

两者最终功能是等同的，用FromEvent也可以做到FromEventPattern的功能（需要更多的代码），反之亦然（更麻烦），我写了几个例子来测试这一点。

FromEvent

两者的签名（选取了通用的一种）

//FromEvent
public static IObservable<TEventArgs> FromEvent<TDelegate, TEventArgs>
              (Func<Action<TEventArgs>, TDelegate> conversion, 
              Action<TDelegate> addHandler,
              Action<TDelegate> removeHandler);

这是通用的Observable创建方法。

TDelegate: 这是事件源的类型
TEventArgs: 这实际上是OnNext中的T参数，而不是TDelegate的参数，OnNext是Action<T>，只能接收一个参数，这个泛类型就指明了Observer创建的OnNext处理函数中需要处理的参数
conversion: 通常来说TDelegate和OnNext签名是不同的，此时需要一个转换函数，将Tdelegate转化为对OnNext的调用
addHandler/removeHandler: 注册和取消注册事件，在Observer调用Subscribe订阅时注册。

实际上可以看出来，Observable将源事件(符合类型TDelegate)类型，转化为对OnNext的调用。
某种程度上来说，可以看做event的+操作(可能会有疑问那为何要rx这一层，observer实际能力是在对事件的过滤处理上，即在事件->事件处理函数的路径上，增加了各种机能）。

FromEventPattern

//FromEventPattern
public static IObservable<EventPattern<TEventArgs>> FromEventPattern<TDelegate, TEventArgs>(Action<TDelegate> addHandler, Action<TDelegate> removeHandler);

该签名和FromEvent基本相同，区别在于少了一个convension转换参数。

这是因为该函数设计即是用于对接.net 标准的EventHandler类型的事件。EventHandler中包含两个参数sender和EventArgs，而这个函数即会将sender和EventArgs两个参数构造为一个EventPattern<EventArgs>类型的参数，传递给OnNext。
也就是说，RX内部帮助我们实现了对于.net 标准的EventHandler类型事件的conversion转换函数。

注意到这里也有TDelegate和TEventArgs两个参数，这帮助我们扩展符合EventHandler类型事件的扩展格式。EventHandler中的EventArgs可以被指定为TEventArgs代表的类型，EventHandler也可以被相应扩展为EventHandler<TEventArgs>类型的事件格式。

示例代码

我基于我的原本需求FileSystemWatcher监控，写了一些示例代码来验证使用方式和差别。

Observable基于FileSystemWatcher的Changed事件，Observer的OnNext实现全部使用相同的打印函数，而通过不同的Observerbal创建方法，达成同样的目的。

最后还增加了一个同时监控Changed和Renamed事件，并且增加我最初防抖动目的的代码。

框架代码如下：

//Create watcher
var watcher = new FileSystemWatcher();
watcher.Path = "d:\test";

//creation of observable goes here
var observable = .....

//subscribe to observable
var d = observable
	.Subscribe(
		a => Console.WriteLine($"Watcher type:{a.ChangeType}, file:{a.Name}"),
		ex => Console.WriteLine("observer error:" + ex.ToString()),
		() => Console.WriteLine("observer complete.")
);			
//Enable watcher
watcher.EnableRaisingEvents = true;

FromEventPattern

由于Changed事件签名符合EventHandler<FileSystemEventArgs>类型，因此通过这种方式可以如下定义observable:

var observable = Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
                            ev => watcher.Changed += ev,
                            ev => watcher.Changed -= ev).Select(x => x.EventArgs);

注意这里最后的Select，它将EventPattern<FileSystemEventArgs>转化为FileSystemEventArgs。

FromEvent

如果通过FromEvent，则需要将EventHandler<FileSystemEventArgs>转化为对OnNext的调用。
如果不通过conversion函数使用FromEvent，则在代码运行时创建代理会失败：

//THIS IS WRONG!
var observable = Observable.FromEvent<FileSystemEventHandler, FileSystemEventArgs>(
                            ev => watcher.Changed += ev,
                            ev => watcher.Changed -= ev);

通过conversion

需要增加一个conversion函数转化

var observable = Observable.FromEvent<FileSystemEventHandler, FileSystemEventArgs>(
                    handler =>
                    {
                        return (object sender, FileSystemEventArgs fa) => handler(fa);
                    }
                    , ev => watcher.Changed += ev,
                            ev => watcher.Changed -= ev);

通过修改源代理类型

稍微麻烦，只是作为展示FromEvent的使用，当签名相同时，可不用conversion重载。

首先定义observable使用的代理类

private delegate void MyWatcherDelegate(FileSystemEventArgs args);

private static event MyWatcherDelegate MyWatcherEvent;

然后将watcher的事件订阅到自定义的MyWatcherEvent事件中.（注意，由于这里使用的是匿名函数，所以无法从Watcher的事件中删除）

watcher.Changed += (sender, args) =>
                {
                    Console.WriteLine("Intermediate watcher handler delegate");
                    if (null != MyWatcherEvent)
                    {
                        Console.WriteLine("Call through custom event");
                        MyWatcherEvent(args);
                    }
                };

最后通过FromEvent创建Observable，这时用于MyWatcherDelegate符合OnNext的签名，因此不需要conversion方法，也可以成功创建。

var observable = Observable.FromEvent<MyWatcherDelegate, FileSystemEventArgs>(
                   ev => MyWatcherEvent += ev,
                   ev => MyWatcherEvent -= ev);

合并Renamed和Changed事件

这里其实和标题关系不大，而是扩展Observable的使用。
通过Merge能使observable同时监听两个签名不同的事件：

var o1 = Observable.FromEventPattern<FileSystemEventHandler, FileSystemEventArgs>(
                            ev => watcher.Changed += ev,
                            ev => watcher.Changed -= ev);
var o2 = Observable.FromEventPattern<RenamedEventHandler, RenamedEventArgs>(
            ev => watcher.Renamed += ev,
            ev => watcher.Renamed -= ev)
    .Select(x => new System.Reactive.EventPattern<FileSystemEventArgs>(x.Sender, x.EventArgs));

var observable = Observable.Merge(o1, o2);