标签: system.reactive

我在使用Reactive Extensions for .NET的WPF应用程序中有以下代码:

public MainWindow()
{
    InitializeComponent();

    var leftButtonDown = Observable.FromEvent<MouseButtonEventArgs>(this, "MouseLeftButtonDown");
    var leftButtonUp = Observable.FromEvent<MouseButtonEventArgs>(this, "MouseLeftButtonUp");

    var moveEvents = Observable.FromEvent<MouseEventArgs>(this, "MouseMove")
        .SkipUntil(leftButtonDown)
        .SkipUntil(leftButtonUp)
        .Repeat()
        .Select(t => t.EventArgs.GetPosition(this));

    moveEvents.Subscribe(point =>
    {
        textBox1.Text = string.Format(string.Format("X: {0}, Y: {1}", point.X, point.Y));
    });
}

当鼠标在此对话框上移动时,内存是否会稳定增加？

读取代码,我希望moveEvents observable会在一段时间后包含大量的MouseEventArgs吗？或者这是以一种我不知道的聪明方式处理的？

我有2个线程,WPF + PIPE.我在管道rx事件上注册了WPF.当使用ObserveOnDispatcher()时,不会调用已注册的处理程序,当删除ObserveOnDispatcher()时,它会在管道线程上调用.有没有人有想法为什么在使用ObserveOnDispatcher()时根本没有调用它？

以下内容未编译VS2010,SL5和Reactive Extensions版本1.1.10605.1(2011-06-05).

'System.Collections.Generic.IEnumerable'不包含'ToObservable'的定义,也没有扩展方法'ToObservable'接受类型为'System.Collections.Generic.IEnumerable'的第一个参数'(你是否缺少using指令)或汇编参考？)

using System;
using System.Collections.Generic;
using System.Linq;
using System.Net;
using System.Windows;
using System.Windows.Controls;
using System.Windows.Documents;
using System.Windows.Input;
using System.Windows.Media;
using System.Windows.Media.Animation;
using System.Windows.Shapes;

public partial class MainPage : UserControl
{
    public MainPage()
    {
        InitializeComponent();
 }


 private void ButtonNormal_Click(object sender, RoutedEventArgs e)
    {
        //MyList.ItemsSource = GetModel();
        var items = new System.Collections.ObjectModel.ObservableCollection<string>();
        MyList.ItemsSource = items;
        foreach (string item in GetModel())
            items.Add(item);
    }

    private void ButtonRx_Click(object sender, RoutedEventArgs e)
    {
        var items = new System.Collections.ObjectModel.ObservableCollection<string>();
        MyList.ItemsSource = items;

        IObservable<string> observable = …

我有一个方法,它Progress<T>作为参数,并在内部报告进展.

因为进度可能在工作线程上快速发生(例如每秒数百次),我需要缓冲它,然后用接收的记录更新视图模型,例如每半秒一次.

在过去我使用过,Observable.FromEventPattern().Buffer(TimeSpan)所以我看到如果我将Progress<T>报告包装到一个事件中,就可以使用相同的Rx机制.然而,这似乎是一种矫枉过正.有更优雅的解决方案吗？

我也看过TPL Dataflow,BufferBlock但我不确定它是否支持基于时间的缓冲,例如每半秒一次.

如果有人有例子,请发布.非常感谢.

我想制作一个快速爆发的火码,它会以30毫秒的间隔称我为20次.

这是我到目前为止所尝试的:

Observable.Repeat(20).Delay(TimeSpan.FromMilliseconds(30)).Subscribe() 

这对我来说似乎是合乎逻辑的解决方案,但我的应用程序正在崩溃,我认为它是因为它永远不会停止.

使用UniRx我该如何实现？ https://rx.codeplex.com/

我有一个要求,在10秒的初始延迟之后,我需要SomeMethod每10分钟执行一次,但是在完成之后应该启动10分钟计时器SomeMethod.这是一个粗略的例子:

Start Task 00:00:00
(10 second delay)
SomeMethod executed at 00:00:10 (takes 15 minutes)
(10 minute delay)
SomeMethod executed at 00:25:10 
... and so on.

我知道如何使用TPL来做到这一点.我可以使用Task.Delay启动任务并执行SomeMethod,然后在每次完成(ContinueWith TaskStatus.RanToCompletion)之后,我创建一个新任务并SomeMethod再次执行.

我的问题是,这有可能使用Observable.Timer吗？就像是...

Observable.Timer(TimeSpan.FromSeconds(10), TimeSpan.FromMinutes(10))

这段代码的问题是,如果SomeMethod需要15分钟,我将SomeMethod运行两个不同的实例,这是我不想要的.我希望10分钟计时器在 SomeMethod完成后启动.这可能是使用Observable或应该留在TPL吗？

编辑:忘了提到我想SomeMethod在它自己的线程中运行.

我需要在核心界面中定义返回列表的方法.我的项目很大程度上依赖于使用async/await所以我需要将我的核心引用/接口定义为尽可能异步.我还将EF7用于我的数据访问层.我目前IAsyncEnumerable到处使用.

我目前正在决定是继续使用IAsyncEnumerable还是恢复使用Task<IEnumerable<T>>.IAsyncEnumerable在这一点上似乎很有希望.EF7也在使用它.麻烦的是,我不知道也无法弄清楚如何使用它.网站上几乎没有任何内容告诉任何人如何使用Ix.Net.有一个ToAsyncEnumerable我可以在IEnumerable对象上使用的扩展,但这不会异步做任何事情(或者它？).另一个缺点是给出以下签名:

IAsyncEnumerable GetPersons();

因为这不是返回Task的函数,所以我不能在函数块中使用async/await.

另一方面,我的直觉告诉我,我应该坚持使用Task<IEnumerable<T>>.这当然也有它的问题.EF没有返回此类型的扩展方法.它有一个ToArrayAsync和ToListAsync扩展方法,但当然,这需要你打电话等待方法中,因为Task<T>不是协变的.这可能是一个问题,因为这会产生额外的操作,如果我只是返回Task对象就可以避免这种操作.

我的问题是:我应该继续使用IAsyncEnumerable(首选)还是应该将所有内容更改回Task<IEnumerable<T>>(不是首选)？我也对其他建议持开放态度.

我有一个我想要同时下载的页面的URL列表HttpClient.URL列表可能很大(100或更多!)

我目前有这个代码:

var urls = new List<string>
            {
                @"http:\\www.amazon.com",
                @"http:\\www.bing.com",
                @"http:\\www.facebook.com",
                @"http:\\www.twitter.com",
                @"http:\\www.google.com"
            };

var client = new HttpClient();

var contents = urls
    .ToObservable()
    .SelectMany(uri => client.GetStringAsync(new Uri(uri, UriKind.Absolute)));

contents.Subscribe(Console.WriteLine);

问题是:由于使用的原因SelectMany,几乎同时创建了大量的任务.似乎如果URL列表足够大,很多任务会给出超时(我得到"任务被取消"例外).

所以,我认为应该有一种方法,可能使用某种调度程序,来限制并发任务的数量,在给定时间不允许超过5或6.

通过这种方式,我可以获得并发下载而无需启动太多可能会失速的任务,就像他们现在所做的那样.

如何做到这一点,我不会因为大量的超时任务而饱和？

十分感谢.

假设我写

var gen = Observable.Range(1, 3)
  .SelectMany(x => Observable.Range(1, x));

产生的序列1 1 2 1 2 3如预期.但是如果我写的话

var gen = Observable.Range(1, 4)
  .SelectMany(x => Observable.Range(1, x));

现在产生的序列1 1 2 1 2 1 3 2 3 4不是人们所期望的1 1 2 1 2 3 1 2 3 4.这是为什么？是否SelectMany()做一些多线程合并？

我发现了各种各样的SO问题,但无法找出F#解决方案.我需要阻止等待事件向我发射以检查它返回的数据.我使用Rx接收事件3次:

let disposable =
    Observable.take 3 ackNack
    |> Observable.subscribe (
        fun (sender, data) ->
            Console.WriteLine("{0}", data.AckNack)
            Assert.True(data.TotalAckCount > 0u)
    )

我想将结果转换为列表,以便稍后可以通过测试框架(xUnit)进行检查,或等待所有3个事件完成并通过Assert.True.

在继续之前,我将如何等待3场比赛？我可以看到Observable.wait其他消息来源的建议Async.RunSynchronously.