-repeat band-aid

gistfile1.m

describe(@"-repeat", ^{
  __block id<RACSubscribable> subscribable = nil;
  
  beforeEach(^{
    subscribable = [RACSubscribable createSubscribable:^RACDisposable *(id<RACSubscriber> subscriber) {
      [subscriber sendNext:nil];
      [subscriber sendCompleted];
      return nil;
    }];
  });
  
  it(@"shouldn't loop forever", ^{
    // This is going to crash if it fails
    __block RACDisposable *disposable = [[subscribable repeat] subscribeNext:^(id _) {
      [disposable dispose];
    }];
    expect(YES).to.beTruthy();
  });
  
  it(@"should interact properly with take and replay", ^{
    __block NSUInteger receivedCount = 0;
    
    RACReplaySubject *subject = [RACReplaySubject subject];
    [subject sendNext:@(1)];
    [subject sendCompleted];
    [[[subject repeat] take:1] subscribeNext:^(id x) {
      ++receivedCount;
    }];
    
    expect(receivedCount).will.equal(1);
  });
});


- (RACSubscribable *)repeat {
  return [RACSubscribable createSubscribable:^(id<RACSubscriber> subscriber) {
    __block RACDisposable *currentDisposable = nil;
    __block BOOL stopRepeating = NO;
    __block BOOL resubscribing = NO;
    
    __block RACSubscriber *innerObserver = [RACSubscriber subscriberWithNext:^(id x) {
      if (!resubscribing) {
        [subscriber sendNext:x];
      }
    } error:^(NSError *error) {
      if (!resubscribing) {
        [subscriber sendError:error];
      }
    } completed:^{
      currentDisposable = nil;
      resubscribing = YES;
      dispatch_async(dispatch_get_current_queue(), ^{
        if (!stopRepeating) {
          resubscribing = NO;
          currentDisposable = [self subscribe:innerObserver];
        }
      });
    }];
    
    currentDisposable = [self subscribe:innerObserver];
    
    return [RACDisposable disposableWithBlock:^{
      stopRepeating = YES;
      [currentDisposable dispose];
    }];
  }];
}