小贝子编程

从操作创建可取消的IOobservable



我想创建一个实用程序方法，为只在订阅时调用的Action创建一个IObservable！它遵循SubscribeOn(…)指令。以下是我的实现，它基于我可以从中提取的内容http://www.introtorx.com和其他资源，但在一个特定情况下失败：

/// <summary>
/// Makes an observable out of an action. Only at subscription the task will be executed. 
/// </summary>
/// <param name="action">The action.</param>
/// <returns></returns>
public static IObservable<Unit> MakeObservable_2(Action action)
{
return Observable.Create<Unit>(
observer =>
{
return System.Reactive.Concurrency.CurrentThreadScheduler.Instance.Schedule(
() =>
{
try
{
action();
observer.OnNext(Unit.Default);
observer.OnCompleted();
}
catch (Exception ex)
{
observer.OnError(ex);
}
});
});
}

我希望CurrentThreadScheduler的使用将导致SubscribeOn()中给出的Scheduler的用法。此实现适用于.SubscribeOn(TaskPoolScheduler.Default)，但不适用于.SSubscribeOn(Dispatcher.CurrentDispatcher)。能否更改以上实现，使下面的所有单元测试都通过？

[Test]
public void RxActionUtilities_MakeObservableFromAction_WorksAsExpected()
{
ManualResetEvent evt = new ManualResetEvent(false);
// Timeout of this test if sth. goes wrong below
Task.Factory.StartNew(() =>
{
Thread.Sleep(5000);
Console.WriteLine("Test timed out!");
evt.Set();
});
int threadIdOfAction = -42;
int threadIdOfSubscriptionContect = -43;
bool subscriptionWasCalled = false;
Action action = () =>
{
threadIdOfAction = Thread.CurrentThread.ManagedThreadId;
Console.WriteLine("This is an action on thread " + threadIdOfAction);
};
var observable = RxActionUtilities.MakeObservable_2(action);
threadIdOfSubscriptionContect = Thread.CurrentThread.ManagedThreadId;
Console.WriteLine("Before subscription on thread " + threadIdOfSubscriptionContect);
// The next line is the one I want to have working, but the subscription is never executed
observable.SubscribeOn(Dispatcher.CurrentDispatcher).Subscribe(
//observable.Subscribe( // would pass
(unit) =>
{
Console.WriteLine("Subscription: OnNext " + threadIdOfAction + ", " + threadIdOfSubscriptionContect);
subscriptionWasCalled = true;
},
(ex) => evt.Set(), () => evt.Set());
Console.WriteLine("After subscription");
evt.WaitOne();
Assert.AreNotEqual(-42, threadIdOfAction);
Assert.AreNotEqual(-43, threadIdOfSubscriptionContect);
Assert.AreEqual(threadIdOfAction, threadIdOfSubscriptionContect);
Assert.That(subscriptionWasCalled);
}
[Test]
// This test passes with the current implementation
public void RxActionUtilities_MakeObservableFromActionSubscribeOnDifferentThread_WorksAsExpected()
{
ManualResetEvent evt = new ManualResetEvent(false);
// Timeout of this test if sth. goes wrong below
Task.Factory.StartNew(() =>
{
Thread.Sleep(5000);
Console.WriteLine("Test timed out!");
evt.Set();
});
int threadIdOfAction = 42;
int threadIdOfSubscriptionContect = 43;
bool subscriptionWasCalled = false;
Action action = () =>
{
threadIdOfAction = Thread.CurrentThread.ManagedThreadId;
Console.WriteLine("This is an action on thread " + threadIdOfAction);
};
var observable = RxActionUtilities.MakeObservable_2(action);
threadIdOfSubscriptionContect = Thread.CurrentThread.ManagedThreadId;
Console.WriteLine("Before subscription on thread " + threadIdOfSubscriptionContect);
// The next line is the one I want to have working, but the subscription is never executed
observable.SubscribeOn(TaskPoolScheduler.Default).Subscribe(
(unit) =>
{
Console.WriteLine("Subscription: OnNext " + threadIdOfAction + ", " + threadIdOfSubscriptionContect);
subscriptionWasCalled = true;
},
(ex) => evt.Set(), () => evt.Set());
evt.WaitOne();
Console.WriteLine("After subscription");
Assert.AreNotEqual(-42, threadIdOfAction);
Assert.AreNotEqual(-43, threadIdOfSubscriptionContect);
Assert.AreNotEqual(threadIdOfAction, threadIdOfSubscriptionContect);
Assert.That(subscriptionWasCalled);
}

[Test]
public void RxActionUtilities_MakeObservableFromAction_IsCancellable()
{
ManualResetEvent evt = new ManualResetEvent(false);
// Timeout of this test if sth. goes wrong below
Task.Factory.StartNew(() =>
{
Thread.Sleep(5000);
Console.WriteLine("Test timed out!");
evt.Set();
});
int threadIdOfAction = -42;
int threadIdOfSubscriptionContect = -43;
bool subscriptionWasCalled = false;
bool actionTerminated = false;
Action action = () =>
{
threadIdOfAction = Thread.CurrentThread.ManagedThreadId;
for (int i = 0; i < 10; ++i)
{
Console.WriteLine("Some action #" + i);
Thread.Sleep(200);
}
actionTerminated = true;
evt.Set();
};
var observable = RxActionUtilities.MakeObservable_2(action);
threadIdOfSubscriptionContect = Thread.CurrentThread.ManagedThreadId;
Console.WriteLine("Before subscription on thread " + threadIdOfSubscriptionContect);
var subscription =
observable.SubscribeOn(TaskPoolScheduler.Default).Subscribe(
(unit) =>
{
Console.WriteLine("Subscription: OnNext " + threadIdOfAction + ", " + threadIdOfSubscriptionContect);
subscriptionWasCalled = true;
},
(ex) => evt.Set(), () => evt.Set());
Console.WriteLine("After subscription");
Thread.Sleep(1000);
Console.WriteLine("Killing subscription ...");
subscription.Dispose();
Console.WriteLine("... done.");
evt.WaitOne();
Assert.IsFalse(actionTerminated);
Assert.AreNotEqual(-42, threadIdOfAction);
Assert.AreNotEqual(-43, threadIdOfSubscriptionContect);
Assert.AreEqual(threadIdOfAction, threadIdOfSubscriptionContect);
Assert.That(subscriptionWasCalled);
}

更新

为了回应李详尽的回答，我再次尝试并重新表述我的问题。IIUC我们可以总结

  • 无法停止已启动的操作
  • 我完全误解了Dispatcher.CurrentDispatcher及其工作方式：AFAICS它永远不应该用作SubscribeOn()的参数，而应该仅用作ObserveOn的参数
  • 我误解了CurrentThreadScheduler

为了创建可取消的内容，我们需要知道取消的操作，例如使用Action<CancellationToken>。这是我的下一次尝试。请告诉我你是否认为这个实现很适合Rx框架，或者我们是否可以再次改进：

public static IObservable<Unit> 
MakeObservable(Action<CancellationToken> action, IScheduler scheduler)
{
return Observable.Create<Unit>(
observer
=>
{
// internally creates a new CancellationTokenSource
var cancel = new CancellationDisposable(); 
var scheduledAction = scheduler.Schedule(() =>
{
try
{
action(cancel.Token);
observer.OnCompleted();
}
catch (Exception ex)
{
observer.OnError(ex);
}
});
// Cancellation before execution of action is performed 
// by disposing scheduledAction
// Cancellation during execution of action is performed 
// by disposing cancel
return new CompositeDisposable(cancel, scheduledAction);
});
}

如果你在做：我不知道如何使用TestSchedulers:来测试

[Test]
public void MakeObservableFromCancelableAction_CancellationTakesPlaceWithTrueThread()
{
var scheduler = NewThreadScheduler.Default;
Action<CancellationToken> action =
(cancellationToken) =>
{
for (int i = 0; i < 10; ++i)
{
Console.WriteLine("Some action #" + i);
if (cancellationToken.IsCancellationRequested)
{
break;
}
Thread.Sleep(20);
// Hoping that the disposal of the subscription stops 
// the loop before we reach i == 4.
Assert.Less(i, 4);
}
};
var observable = RxActionUtilities.MakeObservable(action, scheduler);
var subscription = observable.Subscribe((unit) => { });
Thread.Sleep(60);
subscription.Dispose();
}

我认为你可以让你的代码更简单，你也可以让测试更简单。Rx的美妙之处在于，您应该能够取消所有任务/线程/手动重置事件。此外，我假设您也可以只使用NUnit的[Timeout]属性，而不是您的自定义代码。

不管怎样。。。@Per是对的，Observable。开始就是你想要的。你给它一个Action和一个IScheduler，这似乎正是你想要的。

[Test]
public void Run_Action_as_IOb_on_scheduler_with_ObStart()
{
var scheduler = new TestScheduler();
var flag = false;
Action action = () => { flag = true; };
var subscription = Observable.Start(action, scheduler)
.Subscribe();
Assert.IsFalse(flag);
scheduler.AdvanceBy(1);
Assert.IsTrue(flag);
subscription.Dispose(); //Not required as the sequence will have completed and then auto-detached.
}

然而，你可能会注意到它确实有一些奇怪的行为(至少在我这台电脑上的V1中)。具体来说，Observable.Start只会立即运行Action，而不会真正等待订阅可观察序列。因此，调用subscribe，然后在执行操作之前处理订阅是无效的。嗯。

[Test]
public void Run_Action_as_IOb_on_scheduler_with_ObStart_dispose()
{
var scheduler = new TestScheduler();
var flag = false;
Action action = () => { flag = true; };
var subscription = Observable.Start(action, scheduler).Subscribe();

Assert.IsFalse(flag);
subscription.Dispose();
scheduler.AdvanceBy(1);
Assert.IsFalse(flag);   //FAILS. Oh no! this is true!
}
[Test]
public void Run_Action_as_IOb_on_scheduler_with_ObStart_no_subscribe()
{
var scheduler = new TestScheduler();
var flag = false;
Action action = () => { flag = true; };
Observable.Start(action, scheduler);
//Note the lack of subscribe?!
Assert.IsFalse(flag);
scheduler.AdvanceBy(1);
Assert.IsFalse(flag);//FAILS. Oh no! this is true!
}

然而，我们可以遵循您使用Observable.Create的路径。然而，您已经非常接近了，您只需要在Create委托中进行任何调度。只要相信Rx会为你做这件事。

[Test]
public void Run_Action_as_IOb_on_scheduler_with_ObCreate()
{
var scheduler = new TestScheduler();
var flag = false;
Action action = () => { flag = true; };
var subscription = Observable.Create<Unit>(observer =>
{
try
{
action();
observer.OnNext(Unit.Default);
observer.OnCompleted();
}
catch (Exception ex)
{
observer.OnError(ex);
}
return Disposable.Empty;
})
.SubscribeOn(scheduler)
.Subscribe();   //Without subscribe, the action wont run.
Assert.IsFalse(flag);
scheduler.AdvanceBy(1);
Assert.IsTrue(flag);
subscription.Dispose(); //Not required as the sequence will have completed and then auto-detached.
}
[Test]
public void Run_Action_as_IOb_on_scheduler_with_ObCreate_dispose()
{
var scheduler = new TestScheduler();
var flag = false;
Action action = () => { flag = true; };
var subscription = Observable.Create<Unit>(observer =>
{
try
{
action();
observer.OnNext(Unit.Default);
observer.OnCompleted();
}
catch (Exception ex)
{
observer.OnError(ex);
}
return Disposable.Empty;
})
.SubscribeOn(scheduler)
.Subscribe();   //Without subscribe, the action wont run.
Assert.IsFalse(flag);
subscription.Dispose();
scheduler.AdvanceBy(1);
Assert.IsFalse(flag);   //Subscription was disposed before the scheduler was able to run, so the action did not run.
}

如果你希望能够在处理过程中取消实际操作，那么你需要做一些比这更高级的事情。

最后的实现很简单：

public static class RxActionUtilities
{
/// <summary>
/// Makes an observable out of an action. Only at subscription the task will be executed. 
/// </summary>
/// <param name="action">The action.</param>
/// <returns></returns>
/// <example>
/// <code>
/// <![CDATA[
/// RxActionUtilities.MakeObservable_3(myAction)
///                  .SubscribeOn(_schedulerProvider.TaskPoolScheduler)
///                  .Subscribe(....);
/// 
/// ]]>
/// </code>
/// </example>
public static IObservable<Unit> MakeObservable_3(Action action)
{
return Observable.Create<Unit>(observer =>
{
try
{
action();
observer.OnNext(Unit.Default);
observer.OnCompleted();
}
catch (Exception ex)
{
observer.OnError(ex);
}
return Disposable.Empty;
});
}
}

我希望这能有所帮助。

编辑：W.r.t向您介绍Dispatcher在单元测试中的使用情况。我认为，在应用另一层(Rx)来增加混乱之前，首先你应该试着了解它是如何工作的。Rx在WPF中编码时给我带来的一个关键好处是通过调度器抽象Dispatcher。它使我能够轻松地在WPF中测试并发性。例如，这里的这个简单测试失败了：

[Test, Timeout(2000)]
public void DispatcherFail()
{
var wasRun = false;
Action MyAction = () =>
{
Console.WriteLine("Running...");
wasRun = true;
Console.WriteLine("Run.");
};
Dispatcher.CurrentDispatcher.BeginInvoke(MyAction);
Assert.IsTrue(wasRun);
}

如果你运行这个，你会注意到控制台上甚至没有打印任何内容，所以我们没有比赛条件，操作就是永远不会运行。原因是调度程序没有启动它的消息循环。为了纠正这个测试，我们必须用混乱的基础设施代码来填充它。

[Test, Timeout(2000)]
public void Testing_with_Dispatcher_BeginInvoke()
{
var frame = new DispatcherFrame();  //1 - The Message loop
var wasRun = false;
Action MyAction = () =>
{
Console.WriteLine("Running...");
wasRun = true;
Console.WriteLine("Run.");
frame.Continue = false;         //2 - Stop the message loop, else we hang forever
};
Dispatcher.CurrentDispatcher.BeginInvoke(MyAction);
Dispatcher.PushFrame(frame);        //3 - Start the message loop
Assert.IsTrue(wasRun);
}

因此，我们显然不想对WPF中所有需要并发性的测试都这样做。尝试注入帧将是一场噩梦。继续=错误地采取我们无法控制的行动。幸运的是，IScheudler通过它的Schedule方法公开了我们所需要的一切。

下一个CurrentThreadScheduler应该被认为是一个蹦床，而不是一个SynchronizationContext(我认为你认为是这样)。

我认为Observable.Start就是您想要的。http://msdn.microsoft.com/en-us/library/system.reactive.linq.observable.start(v=vs.103).aspx

相关内容

  • 没有找到相关文章

最新更新


热门标签：

javascript python java c# php android html jquery c++ css ios sql mysql arrays asp.net json python-3.x ruby-on-rails .net sql-server django objective-c excel regex ruby linux ajax iphone xml vba spring asp.net-mvc database wordpress string postgresql wpf windows xcode bash git oracle list vb.net multithreading eclipse algorithm macos powershell visual-studio image forms numpy scala function api selenium