objective-c - 在 dispatch_queue sendComplete 上创建的 RACSubject 没有进入合并信号

Question

当在 DISPATCH_QUEUE_PRIORITY_DEFAULT gcd 队列上运行的 dispatch_async block 中时:我创建了两个 RACSubject 对象，使用 RACSignal merge: 然后订阅完成。然后，出于此测试的目的(并在我的实际代码中复制该场景)，我对它们都发送了 sendComplete。合并信号完成订阅永远不会触发。我独立地为主题附加了两个完成订阅，那些确实火了。如果我在主线程而不是 gcd 队列上执行相同的测试，那么它会按预期工作。

有没有办法让它工作，或者我是否必须重构才能让我的所有主题都在主线程上？

#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

所以这是一个相当糟糕的案例，由 GCD 和 RAC 的交互引起。严格来说，没有错误。但它是令人惊讶和奇怪的。我们在 https://github.com/ReactiveCocoa/ReactiveCocoa/blob/1bd47736f306befab64859602dbdea18f7f9a3f6/Documentation/DesignGuidelines.md#subscription-will-always-occur-on-a-scheduler 的设计指南中讨论了这一要求。 .

关键是订阅必须始终发生在已知的调度器上。这是 RAC 内部强制执行的要求。如果您只是使用普通的旧 GCD，则没有已知的调度程序，因此 RAC 必须将订阅异步发送到调度程序。

所以去参加你的测试:

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

实际的订阅是异步发生的，因为没有已知的调度器。订阅最终发生在 -sendCompleted 调用之后，它完全错过了它们。这确实是一种竞争条件，但实际上您可能永远不会看到它成功。

修复方法是尽可能使用 RACScheduler 而不是 GCD。如果需要使用特定的 GCD 队列，可以使用 RACTargetQueueScheduler。例如，您的测试的有效简化版本:

-(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);
}

由于订阅是从调度程序中发生的，subscribeCompleted: 是同步完成的，获取已完成的事件，一切都如您所愿。

如果您不需要使用特定的 GCD 队列，而只想在非主队列上完成它，那么请执行以下操作:

[[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];
}];

我希望这能澄清您所看到的。如果我需要重新措辞，请告诉我。