Adds -[RACSignal amb:].

ReactiveObjC/RACSignal+Operations.h

@@ -208,6 +208,15 @@ extern const NSInteger RACSignalErrorNoMatchingCase;
 /// the returned signal sends `error` immediately.
 + (RACSignal<ValueType> *)merge:(id<NSFastEnumeration>)signals;
 
+/// Emits all of the items from only the first of the receiver or the given signal
+/// to emit an item or notification.
+///
+/// `amb` will pass through the emissions and notifications of exactly one of the receiver
+/// or given signal : the first one that sends a notification to `amb`, either by emitting an
+/// item or sending an onError or onCompleted notification. `amb` will ignore and discard the
+/// emissions and notifications of the other source signals.
+- (RACSignal<ValueType> *)amb:(RACSignal<ValueType> *)signal;
+
 /// Merges the signals sent by the receiver into a flattened signal, but only
 /// subscribes to `maxConcurrent` number of signals at a time. New signals are
 /// queued and subscribed to as other signals complete.

ReactiveObjC/RACSignal+Operations.m

@@ -494,6 +494,65 @@ + (RACSignal *)merge:(id<NSFastEnumeration>)signals {
 		setNameWithFormat:@"+merge: %@", copiedSignals];
 }
 
+- (RACSignal *)amb:(RACSignal *)signal {
+	NSParameterAssert(signal);
+
+	return [[RACSignal
+		createSignal:^(id<RACSubscriber> subscriber) {
+			__block RACDisposable *d = nil;
+			__block BOOL hasChosen = NO;
+			NSRecursiveLock *lock = NSRecursiveLock.new;
+
+			RACDisposable * (^decide)(RACSignal *, RACDisposable *) = ^(RACSignal *s, RACDisposable *disposeMe) {
+				return [s subscribeNext:^(id _Nullable value) {
+					[lock lock];
+
+					if (!hasChosen)	{
+						hasChosen = YES;
+						[disposeMe dispose];
+						[subscriber sendNext:value];
+						d = [s subscribe:subscriber];
+					}
+
+					[lock unlock];
+				} error:^(NSError *error) {
+					[lock lock];
+
+					if (!hasChosen)	{
+						hasChosen = YES;
+						[disposeMe dispose];
+						[subscriber sendError:error];
+					}
+
+					[lock unlock];
+				} completed:^{
+					[lock lock];
+
+					if (!hasChosen)	{
+						hasChosen = YES;
+						[disposeMe dispose];
+						[subscriber sendCompleted];
+					}
+
+					[lock unlock];
+				}];
+			};
+
+			RACSerialDisposable *d1 = [RACSerialDisposable serialDisposableWithDisposable:nil];
+			RACSerialDisposable *d2 = [RACSerialDisposable serialDisposableWithDisposable:nil];
+
+			d1.disposable = decide(self, d2);
+			d2.disposable = decide(signal, d1);
+
+			return [RACDisposable disposableWithBlock:^{
+				[d dispose];
+				[d1 dispose];
+				[d2 dispose];
+			}];
+		}]
+		setNameWithFormat:@"[%@] -amb: %@", self.name, signal];
+}
+
 - (RACSignal *)flatten:(NSUInteger)maxConcurrent {
 	return [[RACSignal createSignal:^(id<RACSubscriber> subscriber) {
 		RACCompoundDisposable *compoundDisposable = [[RACCompoundDisposable alloc] init];

ReactiveObjCTests/RACSignalSpec.m

@@ -1681,6 +1681,96 @@ + (void)configure:(Configuration *)configuration {
 	});
 });
 
+qck_describe(@"-amb:", ^{
+	__block RACSubject *sub1;
+	__block RACSubject *sub2;
+	__block RACSignal *ambed;
+
+	qck_beforeEach(^{
+		sub1 = [RACSubject subject];
+		sub2 = [RACSubject subject];
+		ambed = [sub1 amb:sub2];
+	});
+
+	qck_it(@"should send all values from first signal to emit a value or a notification", ^{
+		__block NSUInteger nextCount = 0;
+		[ambed subscribeNext:^(id _Nullable _) {
+			nextCount++;
+		}];
+
+		[sub1 sendNext:nil];
+		[sub2 sendNext:nil];
+		[sub1 sendNext:nil];
+		[sub2 sendNext:nil];
+
+		expect(@(nextCount)).to(equal(@2));
+	});
+
+	qck_it(@"should complete if the first signal completes", ^{
+		__block BOOL hasCompleted = NO;
+		__block NSUInteger nextCount = 0;
+		[ambed subscribeNext:^(id _Nullable _) {
+			nextCount++;
+		} completed:^{
+			hasCompleted = YES;
+		}];
+
+		[sub1 sendCompleted];
+		[sub2 sendNext:nil];
+
+		expect(@(hasCompleted)).to(beTruthy());
+		expect(@(nextCount)).to(equal(@0));
+	});
+
+	qck_it(@"should error if the first signal errors", ^{
+		__block BOOL hasError = NO;
+		__block NSUInteger nextCount = 0;
+		[ambed subscribeNext:^(id _Nullable _) {
+			nextCount++;
+		} error:^(NSError * _) {
+			hasError = YES;
+		}];
+
+		[sub1 sendError:nil];
+		[sub2 sendNext:nil];
+
+		expect(@(hasError)).to(beTruthy());
+		expect(@(nextCount)).to(equal(@0));
+	});
+
+	qck_it(@"should not complete if the second signal completes", ^{
+		__block BOOL hasCompleted = NO;
+		__block NSUInteger nextCount = 0;
+		[ambed subscribeNext:^(id _Nullable _) {
+			nextCount++;
+		} completed:^{
+			hasCompleted = YES;
+		}];
+
+		[sub1 sendNext:nil];
+		[sub2 sendCompleted];
+
+		expect(@(hasCompleted)).to(beFalsy());
+		expect(@(nextCount)).to(equal(@1));
+	});
+
+	qck_it(@"should not error if the second signal errors", ^{
+		__block BOOL hasError = NO;
+		__block NSUInteger nextCount = 0;
+		[ambed subscribeNext:^(id _Nullable _) {
+			nextCount++;
+		} error:^(NSError * _) {
+			hasError = YES;
+		}];
+
+		[sub1 sendNext:nil];
+		[sub2 sendError:nil];
+
+		expect(@(hasError)).to(beFalsy());
+		expect(@(nextCount)).to(equal(@1));
+	});
+});
+
 qck_describe(@"-flatten:", ^{
 	__block BOOL subscribedTo1 = NO;
 	__block BOOL subscribedTo2 = NO;