gistfile1.txt

- (RACSignal *)localUpdateSignal
{
    return [[self localBundleUpdateSignal] flattenMap:^RACStream *(NSString *path)
    {
        return [self exerciseUpdateSignalWithBundlePath:path];
    }];
}

- (RACSignal *)localBundleUpdateSignal
{
    RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber)
    {
        NSArray *bundledExercises = [[NSBundle mainBundle] pathsForResourcesOfType:@"zip" inDirectory:nil];
        
        for (NSString *exercise in bundledExercises)
        {
            [subscriber sendNext:exercise];
        }
        
        [subscriber sendCompleted];
        
        return nil;
    }];
    
    return signal;
}

/*!
 *  Returns reactive exercise extraction chain
 *
 *  @param path extracts, reads the data and updates database chain
 *
 *  @return signal
 */
- (RACSignal *)exerciseUpdateSignalWithBundlePath:(NSString *)path
{
    RACSignal *signal = [[[self exerciseExtractSignalWithBundlePath:path] flattenMap:^RACStream *(NSString *path)
    {
        return [self exerciseReadSignalForPath:path];
    }] flattenMap:^RACStream *(RACTuple* result) {
        return [self exerciseUpdateSignalForData:result.first path:result.second];
    }];
    
    return signal;
}

/*!
 *  Extracts archive at path and sends destination path on next event
 *
 *  @param path to extract from
 *
 *  @return signal
 */
- (RACSignal *)exerciseExtractSignalWithBundlePath:(NSString *)path
{
    RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber)
    {
        //
        // Unzip and notify
        //
        NSString *destination = [self destinationPathForArchivePath:path];
        
        [Main unzipFileAtPath:path toDestination:[self destinationWithArchiveDestination:destination] progressHandler:nil completionHandler:^(NSString *destinationPath, BOOL succeeded, NSError *error)
        {
            if (succeeded)
            {
                [subscriber sendNext:destination];
                [subscriber sendCompleted];
            }
            else
            {
                [subscriber sendError:error];
            }
        }];
        
        return nil;
    }];

    return signal;
}

/*!
 *  Signal will read and verify the path, if it contains format for exercise. On success, it will return a RACTuple,
 *  containing directory path and dictionary of exercise data.
 *
 *  @param path to directory with exercise data.
 *
 *  @return signal
 */
- (RACSignal *)exerciseReadSignalForPath:(NSString *)path
{
    RACSignal *signal = [RACSignal createSignal:^RACDisposable *(id<RACSubscriber> subscriber) {
        NSFileManager* fileManager = [NSFileManager defaultManager];
        
        NSError *error;
        
        NSArray* files = [fileManager contentsOfDirectoryAtPath:[self destinationWithArchiveDestination:path] error:&error];
        
        //
        // Check for directory error
        //
        
        if (error)
        {
            [subscriber sendError:error];
            return nil;
        }
        
        //
        // Search for content property list
        //
        
        NSString *contentPath = nil;
        
        for (NSString *file in files)
        {
            if ([file.lastPathComponent isEqualToString:@"Content.plist"] && [file.pathExtension isEqualToString:@"plist"])
            {
                contentPath = file;
            }
        }
        
        if (!contentPath)
        {
            [subscriber sendError:[NSError errorWithDomain:@"parse" code:1 userInfo:nil]];
        }
        else
        {
            NSDictionary *exerciseData = [NSDictionary dictionaryWithContentsOfFile:[NSString stringWithFormat:@"%@/%@", [self destinationWithArchiveDestination:path], contentPath]];
            
            if (exerciseData)
            {
                [subscriber sendNext:RACTuplePack(exerciseData, path)];
                [subscriber sendCompleted];
            }
            else
            {
                [subscriber sendError:[NSError errorWithDomain:@"parse" code:3 userInfo:nil]];
            }
            
        }
        
        return nil;
    }];
    
    return signal;
}

/*!
 *  Signal will create or update exercise with data
 *
 *  @param dictionary data of the exercise
 *  @param path       to exercise located locally
 *
 *  @return signal
 */
- (RACSignal *)exerciseUpdateSignalForData:(NSDictionary *)data path:(NSString *)path
{
    RACSignal *signal = [[[Exercise rac_updateSignal:data] doNext:^(RACTuple *result) {
        Exercise *exercise = result.first;
        
        [[self updatePackageSignalForData:data withExercise:exercise] subscribeCompleted:^{
            NSLog(@"Updated package... %@", [NSThread currentThread]);
        }];
        
    }] flattenMap:^RACStream *(RACTuple *result) {
        Exercise *exercise = result.first;
        
        return [self assetUpdateSignalWithAssets:data[@"assets"] atPath:path forExercise:exercise];
    }];
    
    return signal;
}

- (RACSignal *)updatePackageSignalForData:(NSDictionary *)data withExercise:(Exercise *)exercise
{
    NSDictionary *packageData = nil;
    
    if ([data[@"package"] isKindOfClass:[NSString class]])
    {
        packageData = @{ @"identifier" : data[@"package"] };
    }
    else if ([data[@"package"] isKindOfClass:[NSDictionary class]])
    {
        packageData = data[@"package"];
    }
    
    return [[Package rac_updateSignal:packageData] doNext:^(RACTuple *result) {
        Package* package = result.first;
        
        [MagicalRecord saveWithBlock:^(NSManagedObjectContext *localContext)
        {
            Package *targetPackage = [package MR_inContext:localContext];
            Exercise *targetExercise = [exercise MR_inContext:localContext];
            
            targetExercise.package = targetPackage;
        }];
    }];
}

//
// Code to run this chain
//

[[self localUpdateSignal] subscribeCompleted:nil];

Basically the critical signal (that should be executed on the same thread) is in the last function:

[Package rac_updateSignal]

This is a signal that inserts a new row into the database or retrieves existing row, if a row with same id already exists. The problem I am having is that this gets called simultaneously in certain cases, thus creating two rows with same ids. In this case I must ensure that the signal always executes on the same thread, so only one ID is created. This signal is executed on line 163 with:

 [[self updatePackageSignalForData:data withExercise:exercise] subscribeCompleted:^{
            NSLog(@"Updated package... %@", [NSThread currentThread]);
        }];

I've already tried adding deliverOn with a static background scheduler to this (just before subscribeCompleted), but the thread that is outputted here still happens to be different.

What exactly am I doing wrong?

Hmm, I think you need to put the deliverOn: within the rac_updateSignal function, so it itself performs its work in serial. Another problem could be the MagicalRecord saveWithBlock: – I don't know if that's synchronous or not.

Thanks for the reply, @ashfurrow. I tried that, did not work. The problem was with MagicalRecord, as it created a new instance of NSManagedObjectContext every time. Solved by calling it's synchronous method, so signal does not return until it is saved.