标签: system.reactive

我已经开始研究反应框架了.非常好的东西.但是在查看代码示例时,它让我很困惑.linq语法适用于IQueryable.我认为linq只适用于IEnumerable.关于C#编译器将它的linq基于扩展方法转换的内容是什么？它是否需要一组具有特定接口的方法？

我正在尝试从Silverlight应用程序中删除更多的传统事件处理程序,转而使用大量Rx查询来提供更好,更易于管理的行为抽象.

我需要解决的问题,但不能按照我想要的方式破解它,是让搜索屏幕的行为正常工作.这是非常标准的东西.这是它应该如何表现:

• 我有一个文本框,用户可以在其中输入搜索文本.
• 如果没有文本(或空格只),然后搜索按钮被禁用.
• 当有非空白文本,则搜索按钮被激活.
• 当用户单击搜索时,文本框和搜索按钮都被禁用.
• 当结果返回时,文本框和搜索按钮都被启用.

我有这些observable(通过标准事件的一些扩展方法创建)来使用:

IObservable<IEvent<TextChangedEventArgs>> textBox.TextChangedObservable()
IObservable<IEvent<RoutedEventArgs>> button.ClickObservable()
IObservable<IEvent<LoadingDataEventArgs>> dataSource.LoadingDataObservable()
IObservable<IEvent<LoadedDataEventArgs>> dataSource.LoadedDataObservable()

我现在有这些查询:

IObservable<bool> dataSourceIsBusy =
    dataSource.LoadingDataObservable().Select(x => true)
    .Merge(dataSource.LoadedDataObservable().Select(x => false));

IObservable<string> textBoxText =
    from x in textBox.TextChangedObservable()
    select textBox.Text.Trim();

IObservable<bool> textBoxTextIsValid =
    from text in textBoxText
    let isValid = !String.IsNullOrEmpty(text)
    select isValid; …

最近我决定使用Windows Phone 7的Rx(Reactive Extensions),我遇到了一些奇怪的行为.

例如,我有这段代码:

HttpWebRequest request = (HttpWebRequest)WebRequest.Create("http://whatismyip.org/");
request.Method = "GET";

var x = from c in Observable.FromAsyncPattern<WebResponse>(request.BeginGetResponse, request.EndGetResponse)()
            select c;

WebResponse r = x.First();
Debug.WriteLine(r.ContentType.ToString());

我想弄清楚的是,为什么当我到达LINQ查询时,它会挂起UI并且不会比这更进一步.有任何想法吗？

我想玩RX,我下载了4.0 WPF版本,安装它并开始寻找它的功能示例.我发现了各种代码,我应该很简单,可以复制并粘贴到一个新项目中并开始运行.

我似乎需要参考各种程序集,我找不到任何文档.到目前为止,我有:

Microsoft.Expression.Interactions
C:\Program Files (x86)\Microsoft SDKs\Expression\Blend 3\Interactivity\Libraries\WPF\Microsoft.Expression.Interactions.dll

System.CoreEx
C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2856.0\Net4\System.CoreEx.dll

System.Interactive
C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2856.0\Net4\System.Interactive.dll

System.Reactive
C:\Program Files (x86)\Microsoft Cloud Programmability\Reactive Extensions\v1.0.2856.0\Net4\System.Reactive.dll

System.Windows.Interactivity
C:\Program Files (x86)\Microsoft SDKs\Expression\Blend 3\Interactivity\Libraries\WPF\System.Windows.Interactivity.dll

所以我使用了混合3中的DLL(我已经安装了混合3和4,找不到4版本)然后是看似核心的RX ...

有关这方面的任何建议,指出文档的位置是什么？行为需要交互引用,但实际上这对我来说也是新的.

谢谢伊恩

我正在创建两个(或更多)IObservable<T>所有相同的T.它们是从Task<IEnumerable<T>>哪个可以比其他人更快地回来生成的.我所关心的只是IObservable返回第一个值 - 这是我从那时开始使用的值.

我记得在剑桥参加过Jon Skeet的演讲,他用非常简洁的方式使用TPL完成了这个,但我记不起来了!理想情况下,我会得到一个像这样的方法:

IObservable<T> PickFastestObservable<T>(IEnumerable<IObservable<T>> slowObservables);

但如果我必须直接在任务上完成,我可能会解决一些问题.

我很努力想让自己有点自信,我很自信.

干杯,

我已经订阅了一个推送内容的频率很高的Observable,这些内容来自网络I/O,所以每次推送都来自不同的线程,然后我有一些观察者可能会尝试获取一些内容然后快速取消订阅以确保有没有其他内容传入,所以代码示例如下:

        IDisposable dsp = null;
        dsp = TargetObservable.Subscribe((incomingContent) =>
        {
            if (incomingContent == "something")
            {
                myList.Add(incomingContent);
                dsp.Dispose();
            }
            else
            {
                otherList.Add(incomingContent);
            }
        });

现在,OnNext显然不是线程安全的,意味着当Observer在调用Dispose()之前得到"东西"时,其他内容可能仍然传入并添加到'otherList',即使我放了一个'lock(.. .)'为整个'onNext(...)'.
这不是我们想要的,所以任何想法都要避免这种情况？我可以考虑的一种方法是修改Observable逐个推送内容(通过使用'lock'),然后性能必须伤害很多.谢谢.

有没有一个例子向我展示Observable.Count<TSource> Method实际如何运作？我想出的例子似乎返回一个包含在observable中的计数而不是预期的计数.

例如,我希望1从这里返回:

System.Diagnostics.Debug.WriteLine((Observable.Return<string>("Hello world!")).Count());

将来会1被退回(因为,毕竟它是一个异步序列)？或者我错过了一些基本的东西？在撰写本文时,我实际上假设只要结果被推出,.Count()它将返回结果T并随着时间的推移而增长.真？是.

假设我有IObservable并且我想要一个忽略原始数字的重复数字的可观察量,我该怎么做？我尝试了以下内容

我已经尝试了GroupBy(),但它是一个热门的观察者,它不会起作用.我需要比较的是前一个.

我有第三方API,我没有源代码.我实例化一个回调到这样的事件:

using namespace API; // This is where APIClient lives

namespace TestApiClientUI
{
    public partial class Form1 : Form
    {
        APIClient apiClient = new APICLient();
        apiClient.QuoteUpdated += api_ClientUpdated;

        private void api_ClientUpdated(object sender, string s, double b, double a)
        {
        }
    }
}

如何将其包装到Rx Observable.FromEvent中？

另外,有没有办法做到这一点所以包装副本的开销尽可能少(零拷贝)？

我无法让Observer.onErrorResumeNext按照我的预期行事.

Subscriber orchestratorObserver = new Subscriber<Integer>() {
    @Override
    public void onCompleted() {
        System.out.println("orchestratorObserver."+"onCompleted()");
    }

    @Override
    public void onError(Throwable e) {
        System.out.println("orchestratorObserver."+"onError()"+e.getMessage());
    }

    @Override
    public void onNext(Integer i) {
        System.out.println("orchestratorObserver."+"onNext() : "+i);
    }
};

@Test
public void rxTest() {
    Observable.range(0,5)
            .doOnNext(new Action1<Integer>() {
                @Override
                public void call(Integer integer) {
                    throw new RuntimeException("chain error!!");
                }
            })
            .onErrorResumeNext(Observable.just(-1))
            .subscribe(orchestratorObserver);
}

@Test
public void rxTest1() {

    final Observable<Integer> errorTrainObservable = Observable.defer(
            new Func0<Observable<Integer>>() {
                @Override
                public Observable<Integer> call() {
                    return Observable.error(new RuntimeException("source error"));
                } …