标签: system.reactive

我正在慢慢学习如何使用WPF的.NET Reactive Extensions.有一些初学者的例子说明编写拖放或绘图例程是多么简单,但它们都非常简单.我试图更进一步,对我而言,"正确"方式是什么并不明显.

这些例子都告诉你如何从定义事件流MouseDown,MouseMove以及MouseUp

var mouseDown = from evt in Observable.FromEvent<MouseButtonEventArgs>(..., "MouseDown")
                select evt.EventArgs.GetPosition(...);

var mouseMoves = from evt in Observable.FromEvent<MouseEventArgs>(..., "MouseMove")
                 select evt.EventArgs.GetPosition(...);

var mouseUp = Observable.FromEvent<MouseButtonEventArgs>(..., "MouseUp");

然后如何在a期间轻松完成任务MouseDrag(这将显示从起始拖动点到当前鼠标位置创建的矩形的坐标)

var mouseDrag = from start in mouseDown
                from currentPosition in mouseMoves.TakeUntil(mouseUp)
                select new Rect(Math.Min(start.X, currentPosition.X),
                                Math.Min(start.Y, currentPosition.Y),
                                Math.Abs(start.X - currentPosition.X),
                                Math.Abs(start.Y - currentPosition.Y));

mouseDrag.Subscribe(x =>
             {
                 Info.Text = x.ToString();
             });

我的问题是,在鼠标拖动结束时完成任务的"正确"方法是什么？最初,我以为我可以这样做:

mouseDrag.Subscribe(
     onNext: x =>
             {
                 Info.Text = x.ToString();
             },
     onCompleted: …

我知道有一种简单的方法可以做到这一点 - 但它今晚打败了我......

我想知道两个事件是否在300毫秒之内发生,就像双击一样.

在300毫秒内点击两次左击鼠标 - 我知道这就是反应框架的构建 - 但是如果我能找到一个包含所有扩展操作系统的简单示例的好文档 - Throttle,BufferWithCount,BufferWithTime - 所有这些都不是'为我做这件事......

我试图了解.NET Reactive Extensions,我想知道它们是否可以用于以下场景:

在我的WP7应用程序中,我使用SterlingDatabase来保留应用程序设置.当用户正在修改设置时,我想定期打电话Database.Flush()

因此,在我的Property Set方法中,我想启动Database.Flush()计时器事件,并在5秒内写入数据库.如果写入另一个属性,我想重新启动计时器.

我知道我可以用一个定时器对象做到这一点,调用Start()和Stop(),但我想知道如果我能做到这一点,其中Rx,创建一个Asycn操作,我基本上可以开始与停止,不需使用定时器？

当线程到达时,是否有一种快速记录线程正在等待锁定的方法

lock(x)
{
  //do work on a collection

}

？

如果线程1到达锁定,并且块中没有另一个线程,则应该没有日志.然后,如果线程2到达锁(x)并且线程1仍然在块中,则应该有一个日志.理想的解决方案不需要切换到互斥锁或监视器会增加复杂性.如果有一种快速简便的方法可以使用TPL,PLINQ或RX来做到这一点,那就太棒了.

让我们说:

• 一个待测对象(它内部使用RX);
• 使用RX的测试对象依赖项的数量 - 使用MOQ创建.

问题是:这是一种错误的方法,使用相同的TestScheduler实例来控制测试对象和外部可观察对象的时间吗？

提前致谢

我有两个事件类,叫他们StatusChanged和ValueChanged.我想知道将这些'流'暴露为IObservable.正在实施IObservable<Status>并IObservable<Value>在同一个班级'坏'？是否可能导致我(或我班级的用户)悲伤？

我需要在两个状态之间交替,每个状态具有不同的间隔时间.
我能想到这样做的最好方法是使用Reactive Extensions的Observable.Generate非常棒.

根据我在msdn和其他网站上看到的内容,Observable.Finally()应该在observable"优雅或异常终止"时触发.我正在测试以下代码(在LINQPad中)以查看它是如何工作的,但我根本无法获取.Finall().

var ia = TimeSpan.FromSeconds(1);
var ib = TimeSpan.FromSeconds(.2);
var start = DateTime.UtcNow;
var ct = new CancellationTokenSource();

var o = Observable.Generate(
    true,
//    s => !ct.IsCancellationRequested,
    s => (DateTime.UtcNow-start) < TimeSpan.FromSeconds(3) && !ct.IsCancellationRequested,
    s => !s,
    s => s ? "on" : "off",
    s => s? ib : ia)
//    .TakeUntil(start+TimeSpan.FromSeconds(3))
    .Concat(Observable.Return("end"));


o.Subscribe( s=> s.Dump(), ct.Token);
var t = o.ToTask(ct.Token);


t.ContinueWith(x => x.Dump("done"));
o.Finally(() => "finallY".Dump()); // never gets called?

Thread.Sleep(10000);
ct.Cancel();

如果我使Thread.Sleep 10s,可观察序列完成并且Task.ContinueWith触发,但不是.Finally().

如果我使Thread.Sleep 2s,可观察序列被取消并且Task.ContinueWith再次触发,但不是.Finally(). …

我试图对在IObservable上使用Wait()方法的方法进行单元测试,但是我的测试永远不会完成 - Wait永远不会完成.我的测试包含以下内容:

var scheduler = new TestScheduler();
var input1 = scheduler.CreateColdObservable<List<string>>(
   new Recorded<Notification<List<string>>>(100, Notification.CreateOnNext(new List<string> { "John", "Harry" })),
   new Recorded<Notification<List<string>>>(200, Notification.CreateOnCompleted<List<string>>())
          );

我正在使用Moq通过返回input1来设置我的方法的响应.例如

myObj.Setup(f => f.GetStrings()).Returns(input1);

关于myObj的细节实际上并不重要.我启动调度程序并调用包含Wait的方法(例如我调用的方法中的某个地方)

var results = myObj.GetStrings().Wait();

但这永远不会回来.我怀疑我使用调度程序错误,但我不确定.

关心艾伦

我有独立事件流,与Reactive扩展异步处理.处理程序可能因任何原因而失败,但流继续.

但是,在Rx中,错误发生后,它会自动取消订阅.这有点可配置吗？

例:

async Task<Unit> ActionAsync(int i)
{
    if (i > 1)
        throw new Exception();

    i.Dump();       
    return Unit.Default;
}

void Main()
{
    var sb = new Subject<int>();

    sb.SelectMany(ActionAsync).Subscribe(
        _ => { },
        ex =>
        {
            ex.Dump();
        }
    );


    sb.OnNext(1);
    sb.OnNext(2);
    sb.OnNext(3);
}

我想要以下输出:

• 1
• 例外
• 3

我可以在不尝试/捕捉的情况下实现这一目标ActionAsync吗？

我正在将Reactive Extensions与async/await结合使用,以简化我的套接字协议实现.当特定消息到达时必须执行一些操作(例如,将'pong'发送到每个'ping'消息),并且还有一些方法需要异步等待某些特定响应.以下示例说明了这一点:

private Subject<string> MessageReceived = new Subject<string>();

//this method gets called every time a message is received from socket
internal void OnReceiveMessage(string message)
{
    MessageReceived.OnNext(message);
    ProcessMessage(message);
}

public async Task<string> TestMethod()
{
    var expectedMessage = MessageReceived.Where(x => x.EndsWith("D") && x.EndsWith("F")).FirstOrDefaultAsync();
    await SendMessage("ABC");

    //some code...

    //if response we are waiting for comes before next row, we miss it
    return await expectedMessage;
}

TestMethod()将"ABC"发送到套接字并在接收到例如"DEF"时继续(在此之前可能还有一些其他消息).

这几乎可以,但有一个竞争条件.似乎这段代码不会监听消息,直到return await expectedMessage;这是一个问题,因为有时消息会在此之前到达.