ios - どのようにreactivecoccoaの配列からオブジェクトを送信する(そしてそれぞれを待つ)?

Question

解析からのデータから sqlite のいくつかのテーブルを同期する必要があります。そのためにReactiveCoccoaをセットアップしました。ただし、あるテーブルの同期を待ってから次のテーブルを開始する方法がわかりません。

これは私が今持っているものです：

-(RACSignal *) syncTable:(NSString *)name showProgres:(BOOL)showProgres
{
    @weakify(self);
    NSInteger __block count = 0;

    return [[RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
        PFQuery *query = [PFQuery queryWithClassName:name];
        NSDate *date = [self versionForTable:name];

        if (![name isEqualToString:@"Company"]) {
            [query whereKey:@"x_company" equalTo:[self currentCompany]];
            [query whereKey:@"updatedAt" greaterThan:date];
        } else {
            [query whereKey:@"objectId" equalTo:[self currentCompany].objectId];
        }

        NSArray *objects = [query findObjects];

        if (showProgres) {
            [SVProgressHUD showWithStatus:[NSString stringWithFormat:LOC_SYNC_TABLE, name]];
        }

        for (PFObject *object in objects) {
            [subscriber sendNext:object]; // <-- This is wrong!

            count++;
            if (showProgres) {
                [SVProgressHUD showProgress:count / [objects count] status:name];
            }
        }

        [subscriber sendCompleted];
        return nil;
    }] flattenMap:^RACStream *(PFObject *object) {
        @strongify(self);

        return [self syncRecords:name data:object];
    }];
}

各 PFObject を送信する方法と、進行状況を行う方法がわかりません。

PD: これは、提案された変更を含むコードです。

-(RACSignal *) syncRecords:(NSString *)table data:(PFObject *)data
{
    @weakify(self);

    return [RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
        /* STUFF */

        if (![db saveInternal:rs]) {
            NSDictionary *err = @{NSLocalizedDescriptionKey:[rs errorsAsString]};

            [subscriber sendError:[NSError errorWithDomain:LOC_ERROR_SAVE code:0 userInfo:err]];
        } else {
            /* STUFF */
        }        
        [subscriber sendCompleted];

        return nil;
    }];
}

-(RACSignal *) syncTable:(NSString *)name showProgres:(BOOL)showProgres
{
    @weakify(self);
    NSInteger __block count = 0;

    return [[[RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
/* STUFF */
        NSArray *objects = [query findObjects];
        /* If I do this in async way, the concat is not repected*/
/* STUFF */        
        for (PFObject *object in objects) {
            [subscriber sendNext:object];
/* STUFF */
        }
        return nil;
    }] map:^RACStream *(PFObject *object) {
        @strongify(self);

        return [self syncRecords:name data:object];
    }] concat];
}

-(RACSignal *) fullSync
{
    RACSignal *fullSync = [[[[[RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
/* STUFF */
        for (NSString *table in self.tablesFixed) {
            [subscriber sendNext:table];
        }

        [subscriber sendCompleted];

        return nil;
    }] map:^RACStream *(NSString *table) {
        @strongify(self);

        return [self syncTable:table showProgres:YES];
    }] doError:^(NSError *error) {
        DDLogError(@"%@", error);

        [[Db currentDb] rollbackTransaction];
    }] doCompleted:^{
        [[Db currentDb] commitTransaction];
/* STUFF */
    }] concat];


    return fullSync;
}

-(RACSignal *) autoLogin:(PFUser *)user
{
    @weakify(self);

    return [[RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
/* STUFF */
        [subscriber sendNext:user];

        [subscriber sendCompleted];

        return nil;
    }] flattenMap:^RACStream *(id user) {
        @strongify(self);
        return [self fullSync];
    }];
}

-(RACSignal *) login:(NSString *)email pwd:(NSString *)pwd
{
    DDLogInfo(@"Login user %@", email);

    @weakify(self);

    RACSignal *login = [[RACSignal createSignal:^ RACDisposable *(id<RACSubscriber> subscriber)
    {
        [PFUser logInWithUsernameInBackground:email password:pwd block:^(PFUser *user, NSError *error) {            
            if (error) {
                [subscriber sendError:error];
            } else {
                [subscriber sendNext:user];

                [subscriber sendCompleted];
            }
        }];

        return nil;
    }] flattenMap:^RACStream *(id user) {
       @strongify(self);

       return [self autoLogin:user];
    }];

    return login;
}