Дроссель только в том случае, если встречается определенное условие

Question

У меня есть наблюдаемое, что я подписываюсь. Этот obsevable будет возвращать объект, у которого есть свойство ActivationType, которое может быть установлено несколько раз.

То, что я пытаюсь достичь, - это регистрировать сообщение всякий раз, когда для параметра ActivationType установлено значение «Type1». Однако, если для параметра ActivationType установлено значение «Тип2», запишите сообщение только один раз и подождите 30 секунд, прежде чем регистрировать снова, если ActivationType - «Тип2».

Так что, если у меня есть:

myObservable 
    .Where(o => o.ActivationType == "Type1" || o.ActivationType == "Type2") //listen for types 1 and 2 
    .Throttle() // ??? somehow only throttle if we are currently looking at Type2 
    .Subscribe(Log); //log some stuff 

Я считаю, что дроссельная заслонка() является то, что я ищу, но я не знаю, как вызвать его условно.

Любые предложения?

Answer 1

Ах, идеальный случай для почти невозможно для понимания Window оператора!

EDIT: я отправляю эту ссылку, как дюжину раз в месяц, я клянусь - лучше всего для чтения через я видел в Window, Join, Buffer, GroupJoin и т.д. операторов:

Lee Campbell: Rx Part 9–Join, Window, Buffer and Group Join

var source = new Subject<Thing>(); 

var feed = source.Publish().RefCount(); 
var ofType1 = feed.Where(t => t.ActivationType == "Type1"); 
var ofType2 = feed 
    // only window the type2s 
    .Where(t => t.ActivationType == "Type2") 
    // our "end window selector" will be a tick 30s off from start 
    .Window(() => Observable.Timer(TimeSpan.FromSeconds(30))) 
    // we want the first one in each window... 
    .Select(lst => lst.Take(1)) 
    // moosh them all back together 
    .Merge(); 

    // We want all "type 1s" and the buffered outputs of "type 2s" 
    var query = ofType1.Merge(ofType2); 

    // Let's set up a fake stream of data 
    var running = true; 
    var feeder = Task.Factory.StartNew(
     () => { 
     // until we say stop... 
     while(running) 
     { 
      // pump new Things into the stream every 500ms 
      source.OnNext(new Thing()); 
      Thread.Sleep(500); 
     } 
    }); 

    using(query.Subscribe(Console.WriteLine)) 
    {    
     // Block until we hit enter so we can see the live output 
     // from the above subscribe 
     Console.ReadLine(); 
     // Shutdown our fake feeder 
     running = false; 
     feeder.Wait(); 
    } 

Answer 2

Почему бы просто не использовать два потока?

var baseStream = myObservable.Publish().RefCount(); // evaluate once 
var type1 = baseStream.Where(o => o.ActivationType == "Type1"); 
var type2 = baseStream.Where(o => o.ActivationType == "Type2").Throttle(TimeSpan.FromSeconds(30)); 

type1.Merge(type2).Subscribe(Log);