如何使用 Rx 通过异步 WCF 服务轮询图像

Question

我有一个异步 WCF 服务，它接受“URI”并返回图像（作为流）。

我想要做的是：

• 确保存在有效的 WCF 通道，如果没有创建它
• 则进行异步服务调用
• 成功时将图像保存到成员变量
• 如果出现异常，则关闭通道
• 无论失败还是成功，请等待 200ms 然后重新开始（永远循环或直到取消）

到目前为止，我已经想出了这个怪物：

    private void PollImage(string imageUri)
    {
        const int pollingHertz = 1;
        const int millisecondsTimeout = 1000 / pollingHertz;
        Thread.Sleep(millisecondsTimeout);

        if (_channel == null)
        {
            _channel = _channelFactory.CreateChannel();
        }

        var getImageFunc = Observable.FromAsyncPattern<string, Stream>
                                  (_channel.BeginGetImage, _channel.EndGetImage);

        getImageFunc(imageUri)
            .Finally(() => PollImage(imageUri))
            .Subscribe(
                stream => UpdateImageStream(imageUri, stream),
                ex =>
                    {
                        Trace.TraceError(ex.ToString());
                        ((ICommunicationObject) _channel).CloseOrAbort();
                        _channel = null;
                    });
    }

我真的很想学习 Rx，但每次我尝试时我都会摸不着头脑。

有人愿意给我一些建议吗？谢谢

Answer 1

我为您提供了一个解决方案，但我建议更改您的 PollImage 方法，使其更像 Rx。

签名应该如下所示：

IObservable<Image> PollImage(string imageUri, TimeSpan gapInterval)

您应该将 PollImage 视为一个可观察工厂，并且在您订阅返回的可观察之前，它实际上不会轮询图像。这种方法的优点是它使取消订阅成为可能 - 您的最后一个要点需要这样做 - 并且它干净地分离了轮询图像的代码和更新局部变量的代码。

因此，对 PollImage 的调用如下所示：

PollImage(imageUri, TimeSpan.FromMilliseconds(200.0))
    .Subscribe(image =>
    {
        /* do save/update images here */
    });

实现如下所示：

private IObservable<Image> PollImage(string imageUri, TimeSpan gapInterval)
{
    Func<Stream, Image> getImageFromStream = st =>
    {
        /* read image from stream here */
    };

    return Observable.Create<Image>(o =>
    {
        if (_channel == null)
        {
            _channel = _channelFactory.CreateChannel();
        }

        var getImageFunc =
            Observable
                .FromAsyncPattern<string, Stream>(
                    _channel.BeginGetImage,
                    _channel.EndGetImage);

        var query =
            from ts in Observable.Timer(gapInterval)
            from stream in getImageFunc(imageUri)
            from img in Observable.Using(
                () => stream,
                st => Observable.Start(
                    () => getImageFromStream(st)))
            select img;

        return query.Do(img => { }, ex =>
        {
            Trace.TraceError(ex.ToString());
            ((ICommunicationObject)_channel).CloseOrAbort();
            _channel = null;
        }).Repeat().Retry().Subscribe(o);                   
    });
}

query observable 等待直到 gapInterval 完成，然后调用WCF函数返回流，然后将流转换为图像。

内部的 return 语句做了很多事情。

首先，它使用 Do 运算符捕获发生的任何异常，并像以前一样重置跟踪和通道。

接下来，它调用 .Repeat() 来有效地重新运行查询，使其在再次调用 Web 服务之前等待 gapInterval。我可以在 query 中使用 Observable.Interval 而不是 Observable.Timer 并删除对 .Repeat() 的调用>，但这意味着对 Web 服务的调用每个 gapInterval 都会启动，而不是在上次完成后等待那么长时间。

接下来，它调用 .Retry() ，如果遇到异常，它会有效地重新启动可观察对象，以便订阅者永远不会看到异常。 Do 运算符捕获错误，因此这是可以的。

最后，它订阅观察者并返回 IDisposable ，允许调用代码取消订阅。

除了实现 getImageFromStream 函数之外，仅此而已。

现在请注意。许多人误解了订阅可观察量的工作原理，这可能导致难以发现错误。

以此为例：

var xs = Observable.Interval(TimeSpan.FromSeconds(1.0));

var s1 = xs.Subscribe(x => { });
var s2 = xs.Subscribe(x => { });

s1 & s2 订阅 xs，但它们不是共享一个计时器，而是各自创建一个计时器。您创建了两个 Observable.Interval 内部运作实例，而不是一个。

现在这是可观察量的正确行为。如果其中一个失败，另一个也不会，因为它们不共享任何内部结构——它们彼此隔离。

但是，在您的代码（以及我的代码）中，您存在潜在的线程问题，因为您在对 PollImage 的多次调用中共享 _channel。如果一个调用失败，它会重置通道，这可能会导致并发调用失败。

我的建议是为每个调用创建一个新通道以防止并发问题。

我希望这有帮助。

Answer 2

这就是我想到的（在一些帮助下！）...仍然不是“完美”，但似乎有效。

正如 @Enigma 所说，我现在已经摆脱了共享 _channel 并将其替换为捕获的本地变量。它有效，但我不明白 Rx enuf 来知道这是否是糟糕/有缺陷的方法。我怀疑至少有一种更清洁的方法。

除此之外，我的主要反对意见是 Do()，我称之为 EnsureChannel... 似乎有点臭。但是...它有效...

哦，我必须在 SelectMany 中包含 _（下划线），否则不会再次调用 GetImage。

    private IDisposable PollImage(string imageUri)
    {
        ICameraServiceAsync channel = _channelFactory.CreateChannel();
        return Observable
            .Timer(TimeSpan.FromSeconds(0.2))
            .Do(_ => { channel = EnsureChannel(channel); })
            .SelectMany(_ =>
                Observable
                .FromAsyncPattern<string, Stream>(channel.BeginGetImage, channel.EndGetImage)(imageUri))
            .Retry()
            .Repeat()
            .Subscribe(stream => UpdateImageStream(imageUri, stream));
    }

    private ICameraServiceAsync EnsureChannel(ICameraServiceAsync channel)
    {
        var icc = channel as ICommunicationObject;
        if (icc != null)
        {
            var communicationState = icc.State; // Copy local for debug inspection
            if (communicationState == CommunicationState.Faulted)
            {
                icc.CloseOrAbort();
                channel = null;
            }
        }
        return channel ?? _channelFactory.CreateChannel();
    }

Answer 3

如果你真的想使用，那么人们已经回答了你的问题，如果你正在寻找替代方法，那么我建议你看看 TPL（任务等对象），它允许你从异步方法模式（你的网络）创建任务对象服务调用），然后使用取消令牌启动任务，以便在一段时间后如果任务未完成，您可以通过调用令牌取消方法来取消它。