如何从操作创建可观察量<T>？

Question

我正在尝试从操作创建一个可观察量，但FromEventPattern和FromEvent语法对我来说很难理解。

这是我的尝试：

Observable
.FromEventPattern<Action<IStreamTrade>, IStreamTrade>(
x => _polygonSocket.TradeReceived += x,
x => _polygonSocket.TradeReceived -= x)
.Select(x => x?.EventArgs)
.IsNotNull()
.Subscribe( /** do stuff **/)
.DisposeWith(Bindings);

我觉得FromEventPattern可能不是这里的正确选择，但我无法获得其他编译。

发生什么：

我收到一个Parameter count mismatch.异常，如下所示：

Parameter count mismatch.
Thread started:  #26
at System.Delegate.CreateDelegate (System.Type type, System.Object firstArgument, System.Reflection.MethodInfo method, System.Boolean throwOnBindFailure, System.Boolean allowClosed) [0x000ee] in /Users/builder/jenkins/workspace/xamarin-macios/xamarin-macios/external/mono/mcs/class/corlib/System/Delegate.cs:230 
at System.Delegate.CreateDelegate (System.Type type, System.Object firstArgument, System.Reflection.MethodInfo method) [0x00000] in /Users/builder/jenkins/workspace/xamarin-macios/xamarin-macios/external/mono/mcs/class/corlib/System/Delegate.cs:296 
at System.Reactive.ReflectionUtils.CreateDelegate[TDelegate] (System.Object o, System.Reflection.MethodInfo method) [0x00000] in <370f6a6bb34048878534065376a195cb>:0 
at System.Reactive.Linq.ObservableImpl.FromEventPattern+Impl`2[TDelegate,TEventArgs].GetHandler (System.Action`1[T] onNext) [0x0003d] in <370f6a6bb34048878534065376a195cb>:0 
at System.Reactive.Linq.ObservableImpl.EventProducer`2+Session[TDelegate,TArgs].Initialize () [0x00023] in <370f6a6bb34048878534065376a195cb>:0 
at System.Reactive.Linq.ObservableImpl.EventProducer`2+Session[TDelegate,TArgs].Connect (System.IObserver`1[T] observer) [0x00033] in <370f6a6bb34048878534065376a195cb>:0 

我在寻找什么：

我想学习如何从Action<IStreamTrade>创建可观察量。我对使用Observable.Create的解决方案持开放态度，但如果可能的话，我更喜欢使用FromEvent或FromEventPattern的解决方案。

一些附加信息：

以下是操作的定义方式：

public event Action<IStreamTrade> TradeReceived;

GitHub 源

以下是他们在示例代码中如何使用它：

sockClient.OnTradeUpdate += HandleTradeUpdate;
// later ...
private void HandleTradeUpdate(ITradeUpdate trade)
{
// removed irrelevant code
}

GitHub 源

Answer 1

尝试使用Observable.FromEvent，它将基于操作的通用 .NET 事件转换为可观察序列：

Observable.FromEvent<IStreamTrade>(
x => _polygonSocket.TradeReceived += x,
x => _polygonSocket.TradeReceived -= x)
.Subscribe( /** do stuff **/)
.DisposeWith(Bindings);

Answer 2

虽然我更喜欢，并使用了维塔利的方法。这是使用Observable.Create的替代方法。

Observable
.Create<IStreamTrade>(
x =>
{
Action<IStreamTrade> aist = st => x.OnNext(st);
_polygonSocket.TradeReceived += aist;
return Disposable.Create(() => _polygonSocket.TradeReceived -= aist);
})
.Subscribe()
.DisposeWith(Bindings);