在 Silverlight 中使用 Rx 进行 WCF 调用不适用于 TakeUntil

Question

我有以下代码来设置我的 Rx 连接：

事件相关定义：

    public class QueryEventArgs : EventArgs
    {
        public SomeParametersType SomeParameters
        {
            get;
            set;
        }

        public object QueryContext
        {
                     get;
            set;
        }
    };

    public delegate void QueryDelegate(object sender, QueryEventArgs e);

    public event QueryDelegate QueryEvent;

初始化：

            queryObservable = Observable.FromEvent<QueryEventArgs>(this, "QueryEvent");
            queryObservable.Subscribe((e) =>
            {
                tbQueryProgress.Text = "Querying... ";
                client.QueryAsync(e.EventArgs.SomeParameters, e.EventArgs.QueryContext);
            });

            queryCompletedObservable = from e in Observable.FromEvent<QueryCompletedEventArgs>(client, "QueryCompleted").TakeUntil(queryObservable) select e;

            queryCompletedObservable.Subscribe((e) =>
                {
                    tbQueryProgress.Text = "Ready";
                    SilverlightClientService_QueryCompleted(e.Sender, e.EventArgs);
                },
                (Exception ex) =>
                {
                    SetError("Query error: " + ex);
                }
            );

“客户端”是 WCF 客户端，其他所有内容都是不言自明的。

“TakeUntil”的作用是阻止用户在当前正在运行的查询中进行新查询时踩自己的脚。但是，虽然如果我删除“TakeUntil”子句，代码可以工作，但如果我将其放入，则查询永远不会完成。

这是正确的模式吗？如果是这样，我做错了什么吗？

干杯， -蒂姆

Answer 1

TakeUntil 在从参数接收到值时终止订阅，因此您的第一个 queryObservable 会启动查询，但也会终止对完整事件的订阅。

更简单的解决方案是围绕实际查询设置一个 IObservable，然后使用 Switch 确保一次仅运行一个查询。

public static class ClientExtensions
{
    public static IObservable<QueryCompletedEventArgs> QueryObservable(
        this QueryClient client,
        object[] someParameters, object queryContext)
    {
       return Observable.CreateWithDisposable<QueryCompletedEventArgs>(observer =>
       {
            var subscription = Observable.FromEvent<QueryCompletedEventArgs>(
                    h => client.QueryCompleted += h,
                    h => client.QueryCompleted -= h
                )
                .Subscribe(observer);

            client.QueryAsync(someParameters, queryContext);

            return new CompositeDisposable(
                subscription,
                Disposable.Create(() => client.Abort())
            );
        });
    }
}

然后您可以执行以下操作：

queryObservable = Observable.FromEvent<QueryEventArgs>(this, "QueryEvent");

queryObservable
    .Select(query => client.QueryObservable(
        query.EventArgs.SomeParameters, 
        query.EventArgs.QueryContext
    ))
    .Switch()
    .Subscribe(queryComplete =>
    {
        tbQueryProgress.Text = "Ready";
        // ... etc
    });

这将设置一个连续流，其中每个“查询”事件都会启动一个查询，该查询会从该查询发出完整的事件。新查询会自动终止前一个查询（如果可能）并开始新的查询。