将可观察订阅从UI移动到域层 - 在哪里/如何观察/订阅？ [英] Moving observable subscription from UI to domain tier - Where/How to observe/subscribe?

问题描述

嗨

我有一个griddataview（下面的gridRaces）坐在一个winform上，成功填充 - 从一个可观察到的

I have a griddataview (gridRaces below) sitting on a winform which gets populated - successfully - from an observable via the following

Data.Subscriptions().
       TodaysRaces().
       SubscribeOn(SynchronizationContext.Current).
       Subscribe(races => { gridRaces.DataSource = new BindingList<RaceOverview>(races.ToList()); });

目前它是一切正常，但我在UI层做得太多了。 RaceOverview实际上是一个DTO。真的，我想要检索并将此DTO转换为域层中的域对象（RaceSummary），而不是在UI
层中处理它。 （基础数据来自网络服务）。

At the moment it is all working but I am doing too much in the UI tier. The RaceOverview is really a DTO. Really, I am wanting to retrieve and transform this DTO into a domain object (RaceSummary) in the domain tier rather than handling it all in the UI tier. (The underlying data is coming from a webservice).

理想情况下，我希望同时公开域对象和可观察的域对象。

Ideally I'd like to expose the domain objects both synchronously and as an observable.

a）强制observable完成/阻止的最佳方法是什么，以便可以同步从域层检索数据？ （我目前正在使用OnCompleted  - 虽然当从UI运行时，这似乎永远不会受到影响 - 尽管
来自MSTest - 所以我猜我应该观察或订阅特定的线程？）

a) What is the best way to force the observable to complete/block so that the data can be retrieved from the domain tier synchronously? (I am currently using  OnCompleted  - though when running from the UI this never seems to get hit - though it does from within MSTest - so I'm guessing I should be observing or subscribing on a specific thread?)

b）将数据检索移动到域层时有什么线程问题。我理解当引用UI层中的域层订阅时，我需要订阅当前的同步上下文（如上面的代码示例），但
我不确定是否有任何其他ObserveOn / SubscribeOn此方案中域层中可能需要的问题。

b) What are the threading issues when moving the data retrieval into the domain tier. I understand I'll need to SubscribeOn to the current sync context when referencing the subscription to the domain tier in the UI tier (as in the code sample above), but I'm not sure of any other ObserveOn/SubscribeOn issues which may be required in the domain tier in this scenario.

c）此外，我想知道线程和尝试同步访问数据时是否会有不同的含义。

c) Further I wondered if there would be different implications wrt threading and when trying to access the data synchronously.

对不起，如果这有点笼统。如果我写的东西无法理解，请询问澄清。

Sorry if this is a bit general. Please ask for clarification if there are things which aren't intelligible from what I've written.

提前Thx

S

PS这是我在域层中实现这些东西的时间

PS Here is when I am up to wrt implementating the stuff in the domain tier

        public IObservable<IEnumerable<RaceOverview>> TodaysRacesDTO()
        {
            return (Observable.Using<IEnumerable<RaceOverview>, WSGateway>(
                    () => NewConnection(),
                    (c) => c.TodaysRacesSubscription()));                            <=== I have a test which proves this works
        }

        public IEnumerable<RaceSummary> TodaysRacesSync()
        {
            var completed = false;
            var races = Enumerable.Empty<RaceSummary>();
            var rss = TodaysRacesDTO().Subscribe(
                                     rs =>
                                     {
                                         races = from r in rs
                                                 select new RaceSummary(r);
                                     }, 
                                     () => { completed = true; });   <==== this is never getting hit when run from UI but does from MSTEST

            while (!completed) { }

            return races;

        }
        

  &NBSP; &NBSP; public IObservable< IEnumerable< RaceSummary>> TodaysRacesSummary（）
&NBSP; &NBSP; &NBSP; &NBSP; {
&NBSP; &NBSP; &NBSP; &NBSP; &NBSP; &NBSP;返回TodaysRacesDTO（）。SelectMany（rs => 
               {
&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP; VAR种族=（从r在RS
&NBSP;&NBSP;&NBSP; &NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;选择新RaceSummary（R））; <无线电通信/>&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;&NBSP;返回种族;
&NBSP;&NBSP;&NBSP;&NBSP;&NBSP ;         }）;

      public IObservable<IEnumerable<RaceSummary>> TodaysRacesSummary()
        {
            return TodaysRacesDTO().SelectMany(rs => 
                            {
                                var races = (from r in rs
                                             select new RaceSummary(r));
                                return races;
                            });

;最佳尝试但不起作用...不确定如何将其转换为所需的返回类型 

; best attempt but doesn't work ...unsure how to transform this to the required return type 

; error无法将类型'System.IObservable< RaceSummary>'隐式转换为'System.IObservable< System.Collections.Generic.IEnumerable< RaceSummary>>'。

; error Cannot implicitly convert type 'System.IObservable<RaceSummary>' to 'System.IObservable<System.Collections.Generic.IEnumerable<RaceSummary>>'.

;存在显式转换（您是否错过了演员？）   &NBSP; &NBSP;

;An explicit conversion exists (are you missing a cast?)      

}

$


$

推荐答案

Simon，

Simon,

首先我要说如果您需要阻止异步调用，那就是代码气味，您应该将其更改为非阻塞。

First of all I'd say that if you need to block on an async call, that's a code smell that you should probably change to be non-blocking.

您还应该考虑返回任务< T> 来自您的域层，而不是
IObservable< T> ，原因很简单，因为它传达的是您只期望一个值而不是一个值流。  请参阅

Lee的书中的注释和示例就此而言。

You should also perhaps consider returning Task<T> from your domain tier rather than IObservable<T>, simply because it communicates that you're only expecting a single value rather than a stream of values.  See the notes and example in Lee's book in this regard.

如果您确实想要返回 IObservable< T> ，您的服务类仍然只需要一个方法：

If, you do specifically want to return an IObservable<T>, your service class will still only need the one method:

public IObservable<IEnumerable<RaceSummary>> TodaysRacesDTO()
{
	return Observable.Using<IEnumerable<RaceOverview>, WSGateway>(
			() => NewConnection(),
			(c) => c.TodaysRacesSubscription())
			.SubscribeOn(Scheduler.TaskPool)
			.Take(1)
			.Select(r => r.Select(x => new RaceSummary(r)));
}

这会将WS调用的结果映射到  IEnumerable< RaceSummary>，$ b在Observable流上返回的$ b 。我添加了一个 Take（1）底层调用，因此它保证在一个结果之后完成，无论TodaysRacesSubscription（）返回什么。  

This will map the result of the WS call to an IEnumerable<RaceSummary>, which is returned on the Observable stream. I've added a Take(1) underlying call so it guarantees completion after one result, regardless of what TodaysRacesSubscription() returns.  

对于Observable的阻止订阅，您不需要 TodaysRacesSync
方法。  只需在您的  TodaysRacesSummary （）方法中调用 .First（）/。Single（）/。Last（）。  它绝不是理想的，如果一个值永远不会到达，它将导致死锁，但它肯定会阻塞，直到可观察的
收到一个值。  再次
查看Lee的书。

For a blocking subscription to an Observable, you don't need your TodaysRacesSync method.  Just call .First()/.Single()/.Last() on your TodaysRacesSummary() method.  It's by no means ideal, and will result in deadlocks if a value never arrives, but it will certainly block until that observable receives a value.  Once again, see Lee's book.

请务必仔细考虑使用 ObserveOn 和 SubscribeOn 的位置;尽管有相反的意图，轻率使用会导致过多的上下文切换，性能会因此而降低。  在UI层中，您应该只需要
需要使用 ObserveOn ，并且仅当/当您想要安排回Dispatcher时。  不要在域层中使用它 - 您应该在最后一刻调用它。  在您的域层中使用
SubscribeOn ，在调用基础方法时 - 并且仅当该方法尚未运行异步时才使用。

Make sure you consider carefully where you use ObserveOn and SubscribeOn; frivolous use leads to excessive context switching and performance will degrade as a result, despite the opposite intention.  In the UI tier, you should only need to use ObserveOn, and only if/when you want to schedule back onto the Dispatcher.  Don't use it in the domain tier - you should call it at the last possible moment.  Use SubscribeOn once in your domain tier, when making the call to the underlying method - and only if the method isn't already run async.

希望这会有所帮助。

Marcus。

这篇关于将可观察订阅从UI移动到域层 - 在哪里/如何观察/订阅？的文章就介绍到这了，希望我们推荐的答案对大家有所帮助，也希望大家多多支持IT屋！