ReactiveCocoa takeUntil и TakeWhile не посылает последний «следующий»

Question

Рассмотрим следующий фрагмент кода:

- (RACSignal *)startRouting { 
... 
} 

- (RACSignal *)updateRoutingWithSession:(NSString *)session { 
... 
} 

- (RACSignal *)fetchFlights { 
    return [[self startRouting] flattenMap:^RACStream *(NSString *session) { 
     return [[[[self updateRoutingWithSession:session] 
         delay:2.0f] 
         repeat] 
         takeUntilBlock:^BOOL(RACTuple *operationAndResponse) { 
          AFHTTPRequestOperation *operation = [operationAndResponse first]; 
          NSDictionary *response = [operationAndResponse second]; 
          return [operation isCancelled] || 100 == [response[kPercentComplete] intValue]; 
         }]; 
    }]; 
} 

То, что здесь происходит, что startRouting возвращает RACSignal, который отправляет идентификатор сеанса. updateRoutingWithSession: возвращает RACSignal, который отправляет NSDictionary, включая атрибут PercentComplete. Между опросами есть две секунды.

fetchFlights будет работать до updateRoutingWithSession: имеет PercentComplete из 100.

Мой вопрос в том, что самый последний sendNext:, где takeUntilBlock возвращается true, не достигает RACSubscriber.

Что мне не хватает?

Answer 1

Я нашел это в мире RX. Это обычно решается путем слияния двух сигналов. Тот, который берет повторяющийся источник, пока предикат не будет истинным. А другой, который пропускает, когда предикат является истинным.

Это выглядит как этот

BOOL (^finished)(id _) = ^BOOL(id _) { 
    return predicate; // BOOLean here 
} 

// You want a multicast signal, because multiple signals will subscribe to the source. 
// Multicasting it means that you won't get repeated api-requests, in this case. 
RACMulticastConnection *source = [[theSignal repeat] publish]; 

RACSignal *whileNotDone = [source.signal takeUntilBlock:finished]; 
RACSignal *whenDone = [[source.signal skipUntilBlock:finished] take:1]; 
RACSignal *merged = [RACSignal merge:@[whileNotDone, whenDone]]; 

[source connect]; // Needed for a multicast signal to initiate. 

The merged сигнал будет sendNext каждый next в source, включая самого последнего. Затем sendCompleted.

Некоторые ссылки из мира RX:

• "How to complete a Rx Observable depending on a condition in a event"
• "RX Observable.TakeWhile checks condition BEFORE each element but I need to perform the check after"

Answer 2

Чтобы уточнить: ваша проблема в том, что next, который запускает завершение, не отправляется? takeUntilBlock будет распространять следующие сообщения до тех пор, пока предикат не станет НЕТ. (documentation) Поэтому последние next не будут отправлены. Но вы можете подписаться на completion, который должен произойти в этом случае.