RACSubject created in dispatch_queue sendComplete does not make it a combined signal

Question

If in the dispatch_async block running in the gcd queue DISPATCH_QUEUE_PRIORITY_DEFAULT: I create two RACSubjects, use the RACSignal: merge and then sign up for completion. Then, for the purposes of this test (and to repeat the script in my actual code), I send sendComplete to both of them. A signed subscription to end a subscription never works. I attached to the subscription two subscriptions to complete on their own, they shoot. If I do the same test in the main thread instead of the gcd queue, then it works as expected.

Is there a way to do this work, or will I have to reorganize so that all my topics are covered in the main thread?

#import <ReactiveCocoa/ReactiveCocoa.h> @interface rac_signal_testTests: SenTestCase @end @implementation rac_signal_testTests - (void)setUp { [super setUp]; // Set-up code here. } - (void)tearDown { // Tear-down code here. [super tearDown]; } -(void)test_merged_subjects_will_complete_on_main_thread{ RACSubject *subject1 = [[RACSubject subject] setNameWithFormat:@"subject1"]; RACSubject *subject2 = [[RACSubject subject] setNameWithFormat:@"subject2"]; RACSignal *merged = [RACSignal merge:@[subject1, subject2]]; __block BOOL completed_fired = NO; [merged subscribeCompleted:^{ completed_fired = YES; }]; [subject1 sendNext:@"1"]; [subject2 sendNext:@"2"]; [subject1 sendCompleted]; [subject2 sendCompleted]; STAssertTrue(completed_fired, nil); } //test proving that throttling isn't breaking the merged signal (initial hypothesis). -(void)test_merged_subjects_will_complete_if_one_of_them_has_a_throttled_subscriber_on_main_thread{ RACSubject *subject1 = [[RACSubject subject] setNameWithFormat:@"subject1"]; RACSubject *subject2 = [[RACSubject subject] setNameWithFormat:@"subject2"]; __block NSString * hit_subject2_next = nil; [[subject2 throttle:.5] subscribeNext:^(NSString *value){ hit_subject2_next = value; }]; RACSignal *merged = [RACSignal merge:@[subject1, subject2]]; __block BOOL completed_fired = NO; [merged subscribeCompleted:^{ completed_fired = YES; }]; [subject2 sendNext:@"2"]; [subject2 sendCompleted]; [subject1 sendCompleted]; STAssertEqualObjects(@"2", hit_subject2_next, nil); STAssertTrue(completed_fired, nil); } -(void)test_merged_subjects_will_complete_if_on_gcd_queue{ __block BOOL complete = NO; dispatch_queue_t global_default_queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0); dispatch_async(global_default_queue, ^{ RACSubject *subject1 = [[RACSubject subject] setNameWithFormat:@"subject1"]; RACSubject *subject2 = [[RACSubject subject] setNameWithFormat:@"subject2"]; __block NSString * hit_subject2_next = nil; RACScheduler *global_default_scheduler = [RACScheduler schedulerWithQueue:global_default_queue name:@"com.test.global_default"]; RACSignal *sig1 = [subject1 deliverOn:RACScheduler.mainThreadScheduler]; RACSignal *sig2 = [subject2 deliverOn:RACScheduler.mainThreadScheduler]; [sig2 subscribeNext:^(NSString *value){ hit_subject2_next = value; }]; [sig2 subscribeCompleted:^{ NSLog(@"hit sig2 complete"); }]; [sig1 subscribeCompleted:^{ NSLog(@"hit sig1 complete"); }]; RACSignal *merged = [[RACSignal merge:@[sig1, sig2]] deliverOn:RACScheduler.mainThreadScheduler]; [merged subscribeCompleted:^{ complete = YES; }]; [subject2 sendNext:@"2"]; // if we dispatch the send complete calls to the main queue then this code works but that seems like it shoul be unnecessary. // dispatch_async(dispatch_get_main_queue(), ^{ [subject1 sendCompleted]; [subject2 sendCompleted]; // }); }); NSDate *startTime = NSDate.date; do{ [NSRunLoop.mainRunLoop runMode:NSDefaultRunLoopMode beforeDate:[NSDate dateWithTimeIntervalSinceNow:.5]]; }while(!complete && [NSDate.date timeIntervalSinceDate:startTime] <= 10.0); STAssertTrue(complete, nil); } @end 

Answer 1

So, this is a pretty shitty case caused by the interaction of GCD and RAC. Strictly speaking, there is no mistake. But this surprising and strange. We talk about this requirement in the design guidelines https://github.com/ReactiveCocoa/ReactiveCocoa/blob/1bd47736f306befab64859602dbdea18f7f9a3f6/Documentation/DesignGuidelines.md#subscription-will-always-occurleron-occuronon

The key is that the subscription should always be done on a well-known scheduler . This is a requirement that RAC provides internally. If you just use a plain old GCD, there is no known scheduler, so the RAC should send the scheduler subscription asynchronously.

So, to go to your test:

 [merged subscribeCompleted:^{ complete = YES; }]; 

The actual subscription happens asynchronously because there is no known scheduler. After that, the subscription ends after calls to- -sendCompleted , and it completely skips them. This is indeed a race condition, but apparently you will probably never succeed.

The fix is ​​to use the RACScheduler instead of the GCD, if possible. If you need to use a specific GCD queue, you can use the RACTargetQueueScheduler . For example, a working, simplified version of your test:

 -(void)test_merged_subjects_will_complete_if_on_gcd_queue{ __block BOOL complete = NO; dispatch_queue_t global_default_queue = dispatch_get_global_queue(DISPATCH_QUEUE_PRIORITY_DEFAULT, 0); RACScheduler *scheduler = [[RACTargetQueueScheduler alloc] initWithName:@"testScheduler" targetQueue:global_default_queue]; [scheduler schedule:^{ RACSubject *subject1 = [[RACSubject subject] setNameWithFormat:@"subject1"]; RACSubject *subject2 = [[RACSubject subject] setNameWithFormat:@"subject2"]; RACSignal *merged = [RACSignal merge:@[subject1, subject2]]; [merged subscribeCompleted:^{ complete = YES; }]; [subject1 sendCompleted]; [subject2 sendCompleted]; }]; NSDate *startTime = NSDate.date; do{ [NSRunLoop.mainRunLoop runMode:NSDefaultRunLoopMode beforeDate:[NSDate dateWithTimeIntervalSinceNow:.5]]; }while(!complete && [NSDate.date timeIntervalSinceDate:startTime] <= 10.0); STAssertTrue(complete, nil); } 

Since the subscription comes from the scheduler, subscribeCompleted: runs synchronously, receives completed events, and everything behaves as you expected.

If you don’t need to use a specific GCD queue and just want it to be done in a non-primary queue, do the following:

 [[RACScheduler scheduler] schedule:^{ RACSubject *subject1 = [[RACSubject subject] setNameWithFormat:@"subject1"]; RACSubject *subject2 = [[RACSubject subject] setNameWithFormat:@"subject2"]; RACSignal *merged = [RACSignal merge:@[subject1, subject2]]; [merged subscribeCompleted:^{ complete = YES; }]; [subject1 sendCompleted]; [subject2 sendCompleted]; }]; 

I hope this clarifies what you see. Let me know if I need to rewrite something.