Использование реактивной для повторной подписки каждый раз, когда происходит другое событие

Question

Я хочу подписаться на наблюдаемое, но запускать действие можно только один раз, пока не сработает другое IObservable. Каждый раз, когда другие IObservable пожары, я хотел бы быть повторно подписан и запустить мое действие один раз. Следующий код должен выполнить это:

Action<object> act = null;
act = _ => {
    DoThis();
    obj.Observable1.SkipUntil(obj.CompletedObservable).Take(1).Subscribe(act);
};
obj.Observable1.Take(1).Subscribe(act);

Есть ли более правильный способ сделать это с помощью инфраструктуры Rx? Если нет, как я могу обернуть этот шаблон в расширение?

Answer 1

Кажется, вы хотите что-то вроде этого:

first  -----1------------2-------------3--4--------->
            |            |             |  |
second ---------A--B----------C-D-E-----------F---G->
                |             |               |
final  ---------A-------------C---------------F----- >

Для этого должна работать комбинация SelectMany, Take и TakeUntil.Я бы сказал:

var final = from f in first
            from s in second.Take(1).TakeUntil(first)
            select s;

.Take(1) гарантирует, что вы получите только первый предмет после первого (для 1, 2 и 4)..TakeUntil(first) охватывает 3, так что F передается только один раз (до 4).Если у вас есть подходящее имя для этой операции, было бы легко заключить метод, принимающий две наблюдаемые в качестве параметров.

Answer 2

Мне кажется, я понял:

obj
    .CompletedObservable
    .StartWith(/* value of suitable type for CompletedObservable */)
    .Select(x => obj.Observable1.Take(1))
    .Switch()
    .Subscribe(x => DoThis());

Answer 3

Вы пробовали следующее:

obj.CompletedObservable
    .Select(_ => obj.Observable1.Next().First())
    .Subscribe(_ => DoThis());