请注意,像在
EnumWindows
中使用的回调函数与Rx略有不同。具体来说,回调可以通过其返回值向调用方进行通信,但Rx观察者无法这样做。此外,回调可以接收多个参数,但Rx观察者只接收单个值。因此,您需要将多个参数包装成单个对象。
考虑到这一点,使用
Observable.Create
作为
Subject
的替代方法。这样,只有在实际存在观察者时才注册回调,并在该观察者取消订阅时注销它。
对于您使用的同步API的示例,您可能会像这样做。请注意,在此示例中,实际上没有办法在流中间取消注册回调,因为所有操作都在我们能够返回取消订阅可处理对象之前同步发生。
public static IObservable<Foo> WrapFooApi(string arg1, string arg2)
{
return Observable.Create<Foo>(observer =>
{
FooApi.enumerate(arg1, arg2, e =>
{
observer.OnNext(new Foo(e));
return true;
});
observer.OnCompleted();
return Disposable.Empty;
});
}
WrapFooApi("a", "b").Take(1).Subscribe(...);
我们可以通过引入一些异步性来解决无法提前停止的问题,这将给观察者时间获取一个可处理的对象以通知您。我们可以使用
CreateAsync
来获取一个
CancellationToken
,当观察者取消订阅时,它将被取消。并且我们可以在
Task.Run
中运行
FooApi
代码:
public static IObservable<Foo> WrapFooApi(string arg1, string arg2)
{
return Observable.CreateAsync<Foo>(async (observer, ct) =>
{
await Task.Run(() => FooApi.register_callback(arg1, arg2, e =>
{
observer.OnNext(e);
return !ct.IsCancellationRequested;
}));
observer.OnCompleted();
});
}
在传统的异步回调API中,您需要在某个时刻注册并在另一个时刻取消注册。代码可能会像这样:
public static IObservable<Foo> WrapFooApi(string args)
);
return unsubscribe;
});
}
Subject
并在回调中调用OnNext
。 - DirkSubject.OnNext
再使用Subject.OnComplete
吗? - avoIObservable
和IObserver
。调用 OnNext/OnError/OnCompleted 方法向 Subject 的订阅者发出这些命令。它们作为一种从非 Rx 到 Rx 代码的网关。 - Dirk