Merge pull request #478 from ReactiveCocoa/queue-scheduler

Added +schedulerWithQueue:name:

ReactiveCocoaFramework/ReactiveCocoa/RACScheduler.h

@@ -54,6 +54,21 @@ typedef void (^RACSchedulerRecursiveBlock)(void (^reschedule)(void));
 // Invokes +schedulerWithPriority: with RACSchedulerPriorityDefault.
 + (instancetype)scheduler;
 
+// Creates a new scheduler with the given queue and name.
+//
+// Note that the scheduler can only ensure the serial execution of blocks
+// scheduled through this scheduler and not blocks enqueued directly on the
+// given queue, or even blocks scheduled on the same queue through a different
+// scheduler. If the queue allows for concurrent execution, scheduled blocks may
+// run concurrently with blocks directly enqueued.
+//
+// queue - The queue which the scheduler should target. Cannot be NULL.
+// name  - The name for the scheduler, to be used for debug or instrumentation
+//         purposes only. May be nil.
+//
+// Returns the created scheduler.
++ (instancetype)schedulerWithQueue:(dispatch_queue_t)queue name:(NSString *)name;
+
 // The current scheduler. This will only be valid when used from within a
 // -[RACScheduler schedule:] block or when on the main thread.
 + (instancetype)currentScheduler;

ReactiveCocoaFramework/ReactiveCocoa/RACScheduler.m

@@ -75,6 +75,12 @@ + (instancetype)scheduler {
 	return [self schedulerWithPriority:RACSchedulerPriorityDefault];
 }
 
++ (instancetype)schedulerWithQueue:(dispatch_queue_t)queue name:(NSString *)name {
+	NSParameterAssert(queue != NULL);
+
+	return [[RACQueueScheduler alloc] initWithName:name targetQueue:queue];
+}
+
 + (instancetype)subscriptionScheduler {
 	static dispatch_once_t onceToken;
 	static RACScheduler *subscriptionScheduler;

ReactiveCocoaFramework/ReactiveCocoaTests/RACSchedulerSpec.m

@@ -10,6 +10,7 @@
 #import "RACScheduler+Private.h"
 #import "RACDisposable.h"
 #import "EXTScope.h"
+#import <libkern/OSAtomic.h>
 
 // This shouldn't be used directly. Use the `expectCurrentSchedulers` block
 // below instead.
@@ -234,6 +235,44 @@ static void expectCurrentSchedulersInner(NSArray *schedulers, NSMutableArray *cu
 	});
 });
 
+describe(@"+schedulerWithQueue:name:", ^{
+	it(@"should have a valid current scheduler", ^{
+		dispatch_queue_t queue = dispatch_queue_create("test-queue", DISPATCH_QUEUE_SERIAL);
+		RACScheduler *scheduler = [RACScheduler schedulerWithQueue:queue name:@"test-scheduler"];
+		__block RACScheduler *currentScheduler;
+		[scheduler schedule:^{
+			currentScheduler = RACScheduler.currentScheduler;
+		}];
+
+		expect(currentScheduler).will.equal(scheduler);
+
+		dispatch_release(queue);
+	});
+
+	it(@"should schedule blocks FIFO even when given a concurrent queue", ^{
+		dispatch_queue_t queue = dispatch_queue_create("test-queue", DISPATCH_QUEUE_CONCURRENT);
+		RACScheduler *scheduler = [RACScheduler schedulerWithQueue:queue name:@"test-scheduler"];
+		__block volatile int32_t startedCount = 0;
+		__block volatile uint32_t waitInFirst = 1;
+		[scheduler schedule:^{
+			OSAtomicIncrement32Barrier(&startedCount);
+			while (waitInFirst == 1) ;
+		}];
+
+		[scheduler schedule:^{
+			OSAtomicIncrement32Barrier(&startedCount);
+		}];
+
+		expect(startedCount).will.equal(1);
+
+		OSAtomicAnd32Barrier(0, &waitInFirst);
+
+		expect(startedCount).will.equal(2);
+
+		dispatch_release(queue);
+	});
+});
+
 describe(@"+immediateScheduler", ^{
 	it(@"should immediately execute scheduled blocks", ^{
 		__block BOOL executed = NO;