Rx с внешними состояниями

Question

Rx с внешними состояниями?

Итак, в этом примере есть функциональность Rx, объединенная с полным поведением внешнего состояния. Каков наилучший подход с Rx для достижения этого?

Проблемные кодовые места с «updateActive».

public enum Source 
{ 
    Observable1, 
    Observable2, 
} 

// Type is Source.Observable1 
IObservable<Source> observable1; 

// Type is Source.Observable2 
IObservable<Source> observable2; 

var mergedObservables = observable1.Select(x => Source.Observable1) 
    .Merge(observable2.Select(x => Source.Observable2)); 

var updateActive = false; 

mergedObservables.Subscribe(x => 
{ 
    switch (x.Source) 
    { 
     case Source.Observable1: 
     { 
      if (updateActive) 
       break; 

      updateActive = true; 

      // Here is some code which causes that observable2 will get some new values. 
      // (this coud be also on an other thread) 
      // If this is the case, the new value(s) should be ignored. 

      updateActive = false; 
     } 
     break; 

     case Source.Observable2: 
     { 
      if (updateActive) 
       break; 

      updateActive = true; 

      // Here is some code which causes that observable1 will get some new values. 
      // (this coud be also on an other thread) 
      // If this is the case, the new value(s) should be ignored. 

      updateActive = false; 
     } 
     break; 
    } 
}); 

Примечание: Как я могу транспортировать состояние «updateActive» в операторах Rx

Answer 1

Вы могли бы использовать экземпляр Subject<bool> как государственный перевозчик.

Дано:

Subject<bool> isUpdating = new Subject<bool>(); 

Вы будете использовать его в чем-то вроде этого:

var flaggedObservables = mergedObservables 
    .CombineLatest(isUpdating, (obs, flag) => new {obs, IsUpdating = flag}); 

flaggedObservables 
    .Where(data => !data.IsUpdating) 
    .Select(data => data.obs) 
    .DistinctUntilChanged() 
    .Subscribe(
     obs => 
     { 
      isUpdating.OnNext(true); 

      //do some work on obs 

      isUpdating.OnNext(false); 
     }); 

Answer 2

Если вы получили Task, который представляет обновление, это действительно может работать (хотя я не знаю, если я правильно записал вашу желаемую семантику, потому что из вашего начального сообщения кажется, что вы все равно игнорируете все элементы из источника).

public enum Source 
{ 
    Observable1, 
    Observable2, 
} 

// Type is Source.Observable1 
IObservable<Source> observable1; 

// Type is Source.Observable2 
IObservable<Source> observable2; 

var mergedObservables = observable1.Select(x => Source.Observable1) 
    .Merge(observable2.Select(x => Source.Observable2)); 

mergedObservables 
    .Scan(
     new 
     { 
      Value = (Source?)null, 
      CurrentUpdateTask = (Task)null 
     }, 
     (tuple, value) 
     { 
      if ((tuple.CurrentUpdateTask == null) || (tuple.CurrentUpdateTask.IsCompleted)) 
      { 
       // No update running. Start updating. 
       return new 
       { 
        Value = value, 
        CurrentUpdateTask = Update() //Some Task-returning method that does the update. 
       }; 
      } 

      // Update in flight. Ignore value. 
      return new 
      { 
       Value = (Source?)null, 
       tuple.CurrentUpdateTask 
      }; 
     }) 
    .Where(tuple => tuple.Value.HasValue) 
    .Select(tuple.Value.Value) 
    .Subscribe(...); 

Answer 3

Просто добавьте Where пункты к вашему observable1 и observable2. Используйте System.Threading.Interlocked, чтобы обеспечить значение isActive, распространяемое на другие темы. Обратите внимание, что всегда существует условие гонки, в котором значение может быть достигнуто одновременно с одновременным наблюдением. Оба закончили выполнение, хотя и не одновременно. Этот код останавливает только те значения, которые были получены, которые были сгенерированы, пока значение было истинным.

// Тип Source.Observable1 IObservable observable1;

// Type is Source.Observable2 
IObservable<Source> observable2; 

int isActive = 0; 

var mergedObservables = Observable.Merge(
    observable1 
     .Where(t => Interlocked.CompareExchange(ref isActive, 1, 2) == 0) 
     .Select(x => Source.Observable1), 
    observable2 
     .Where(t => Interlocked.CompareExchange(ref isActive, 1, 2) == 0) 
     .Select(x => Source.Observable2)); 

mergedObservables.Subscribe(x => 
{ 
    switch (x.Source) 
    { 
     case Source.Observable1: 
     { 
      Interlocked.Exchange(ref isActive, 1); 

      // Here is some code which causes that observable2 will get some new values. 
      // (this coud be also on an other thread) 
      // If this is the case, the new value(s) should be ignored. 

      Interlocked.Exchange(ref isActive, 0); 
     } 
     break; 

     case Source.Observable2: 
     { 
      Interlocked.Exchange(ref isActive, 1); 

      // Here is some code which causes that observable1 will get some new values. 
      // (this coud be also on an other thread) 
      // If this is the case, the new value(s) should be ignored. 

      Interlocked.Exchange(ref isActive, 0); 
     } 
     break; 
    } 
});