//
//  YSRACSignal.m
//  ReactiveObjC
//
//  Created by Josh Abernathy on 3/15/12.
//  Copyright (c) 2012 GitHub, Inc. All rights reserved.
//

#import "YSRACSignal.h"
#import "YSRACCompoundDisposable.h"
#import "YSRACDisposable.h"
#import "YSRACDynamicSignal.h"
#import "YSRACEmptySignal.h"
#import "YSRACErrorSignal.h"
#import "YSRACMulticastConnection.h"
#import "YSRACReplaySubject.h"
#import "YSRACReturnSignal.h"
#import "YSRACScheduler.h"
#import "YSRACSerialDisposable.h"
#import "YSRACSignal+Operations.h"
#import "YSRACSubject.h"
#import "YSRACSubscriber+Private.h"
#import "YSRACTuple.h"
#import <libkern/OSAtomic.h>

@implementation YSRACSignal

#pragma mark Lifecycle

+ (YSRACSignal *)createSignal:(YSRACDisposable * (^)(id<YSRACSubscriber> subscriber))didSubscribe {
	return [YSRACDynamicSignal createSignal:didSubscribe];
}

+ (YSRACSignal *)error:(NSError *)error {
	return [YSRACErrorSignal error:error];
}

+ (YSRACSignal *)never {
	return [[self createSignal:^ YSRACDisposable * (id<YSRACSubscriber> subscriber) {
		return nil;
	}] setNameWithFormat:@"+never"];
}

+ (YSRACSignal *)startEagerlyWithScheduler:(YSRACScheduler *)scheduler block:(void (^)(id<YSRACSubscriber> subscriber))block {
	NSCParameterAssert(scheduler != nil);
	NSCParameterAssert(block != NULL);

	YSRACSignal *signal = [self startLazilyWithScheduler:scheduler block:block];
	// Subscribe to force the lazy signal to call its block.
	[[signal publish] connect];
	return [signal setNameWithFormat:@"+startEagerlyWithScheduler: %@ block:", scheduler];
}

+ (YSRACSignal *)startLazilyWithScheduler:(YSRACScheduler *)scheduler block:(void (^)(id<YSRACSubscriber> subscriber))block {
	NSCParameterAssert(scheduler != nil);
	NSCParameterAssert(block != NULL);

	YSRACMulticastConnection *connection = [[YSRACSignal
		createSignal:^ id (id<YSRACSubscriber> subscriber) {
			block(subscriber);
			return nil;
		}]
		multicast:[YSRACReplaySubject subject]];
	
	return [[[YSRACSignal
		createSignal:^ id (id<YSRACSubscriber> subscriber) {
			[connection.signal subscribe:subscriber];
			[connection connect];
			return nil;
		}]
		subscribeOn:scheduler]
		setNameWithFormat:@"+startLazilyWithScheduler: %@ block:", scheduler];
}

#pragma mark NSObject

- (NSString *)description {
	return [NSString stringWithFormat:@"<%@: %p> name: %@", self.class, self, self.name];
}

@end

@implementation YSRACSignal (YSRACStream)

+ (YSRACSignal *)empty {
	return [YSRACEmptySignal empty];
}

+ (YSRACSignal *)return:(id)value {
	return [YSRACReturnSignal return:value];
}

- (YSRACSignal *)bind:(YSRACSignalBindBlock (^)(void))block {
	NSCParameterAssert(block != NULL);

	/*
	 * -bind: should:
	 * 
	 * 1. Subscribe to the original signal of values.
	 * 2. Any time the original signal sends a value, transform it using the binding block.
	 * 3. If the binding block returns a signal, subscribe to it, and pass all of its values through to the subscriber as they're received.
	 * 4. If the binding block asks the bind to terminate, complete the _original_ signal.
	 * 5. When _all_ signals complete, send completed to the subscriber.
	 * 
	 * If any signal sends an error at any point, send that to the subscriber.
	 */

	return [[YSRACSignal createSignal:^(id<YSRACSubscriber> subscriber) {
		YSRACSignalBindBlock bindingBlock = block();

		__block volatile int32_t signalCount = 1;   // indicates self

		YSRACCompoundDisposable *compoundDisposable = [YSRACCompoundDisposable compoundDisposable];

		void (^completeSignal)(YSRACDisposable *) = ^(YSRACDisposable *finishedDisposable) {
			if (OSAtomicDecrement32Barrier(&signalCount) == 0) {
				[subscriber sendCompleted];
				[compoundDisposable dispose];
			} else {
				[compoundDisposable removeDisposable:finishedDisposable];
			}
		};

		void (^addSignal)(YSRACSignal *) = ^(YSRACSignal *signal) {
			OSAtomicIncrement32Barrier(&signalCount);

			YSRACSerialDisposable *selfDisposable = [[YSRACSerialDisposable alloc] init];
			[compoundDisposable addDisposable:selfDisposable];

			YSRACDisposable *disposable = [signal subscribeNext:^(id x) {
				[subscriber sendNext:x];
			} error:^(NSError *error) {
				[compoundDisposable dispose];
				[subscriber sendError:error];
			} completed:^{
				@autoreleasepool {
					completeSignal(selfDisposable);
				}
			}];

			selfDisposable.disposable = disposable;
		};

		@autoreleasepool {
			YSRACSerialDisposable *selfDisposable = [[YSRACSerialDisposable alloc] init];
			[compoundDisposable addDisposable:selfDisposable];

			YSRACDisposable *bindingDisposable = [self subscribeNext:^(id x) {
				// Manually check disposal to handle synchronous errors.
				if (compoundDisposable.disposed) return;

				BOOL stop = NO;
				id signal = bindingBlock(x, &stop);

				@autoreleasepool {
					if (signal != nil) addSignal(signal);
					if (signal == nil || stop) {
						[selfDisposable dispose];
						completeSignal(selfDisposable);
					}
				}
			} error:^(NSError *error) {
				[compoundDisposable dispose];
				[subscriber sendError:error];
			} completed:^{
				@autoreleasepool {
					completeSignal(selfDisposable);
				}
			}];

			selfDisposable.disposable = bindingDisposable;
		}

		return compoundDisposable;
	}] setNameWithFormat:@"[%@] -bind:", self.name];
}

- (YSRACSignal *)concat:(YSRACSignal *)signal {
	return [[YSRACSignal createSignal:^(id<YSRACSubscriber> subscriber) {
		YSRACCompoundDisposable *compoundDisposable = [[YSRACCompoundDisposable alloc] init];

		YSRACDisposable *sourceDisposable = [self subscribeNext:^(id x) {
			[subscriber sendNext:x];
		} error:^(NSError *error) {
			[subscriber sendError:error];
		} completed:^{
			YSRACDisposable *concattedDisposable = [signal subscribe:subscriber];
			[compoundDisposable addDisposable:concattedDisposable];
		}];

		[compoundDisposable addDisposable:sourceDisposable];
		return compoundDisposable;
	}] setNameWithFormat:@"[%@] -concat: %@", self.name, signal];
}

- (YSRACSignal *)zipWith:(YSRACSignal *)signal {
	NSCParameterAssert(signal != nil);

	return [[YSRACSignal createSignal:^(id<YSRACSubscriber> subscriber) {
		__block BOOL selfCompleted = NO;
		NSMutableArray *selfValues = [NSMutableArray array];

		__block BOOL otherCompleted = NO;
		NSMutableArray *otherValues = [NSMutableArray array];

		void (^sendCompletedIfNecessary)(void) = ^{
			@synchronized (selfValues) {
				BOOL selfEmpty = (selfCompleted && selfValues.count == 0);
				BOOL otherEmpty = (otherCompleted && otherValues.count == 0);
				if (selfEmpty || otherEmpty) [subscriber sendCompleted];
			}
		};

		void (^sendNext)(void) = ^{
			@synchronized (selfValues) {
				if (selfValues.count == 0) return;
				if (otherValues.count == 0) return;

				YSRACTuple *tuple = YSRACTuplePack(selfValues[0], otherValues[0]);
				[selfValues removeObjectAtIndex:0];
				[otherValues removeObjectAtIndex:0];

				[subscriber sendNext:tuple];
				sendCompletedIfNecessary();
			}
		};

		YSRACDisposable *selfDisposable = [self subscribeNext:^(id x) {
			@synchronized (selfValues) {
				[selfValues addObject:x ?: YSRACTupleNil.tupleNil];
				sendNext();
			}
		} error:^(NSError *error) {
			[subscriber sendError:error];
		} completed:^{
			@synchronized (selfValues) {
				selfCompleted = YES;
				sendCompletedIfNecessary();
			}
		}];

		YSRACDisposable *otherDisposable = [signal subscribeNext:^(id x) {
			@synchronized (selfValues) {
				[otherValues addObject:x ?: YSRACTupleNil.tupleNil];
				sendNext();
			}
		} error:^(NSError *error) {
			[subscriber sendError:error];
		} completed:^{
			@synchronized (selfValues) {
				otherCompleted = YES;
				sendCompletedIfNecessary();
			}
		}];

		return [YSRACDisposable disposableWithBlock:^{
			[selfDisposable dispose];
			[otherDisposable dispose];
		}];
	}] setNameWithFormat:@"[%@] -zipWith: %@", self.name, signal];
}

@end

@implementation YSRACSignal (Subscription)

- (YSRACDisposable *)subscribe:(id<YSRACSubscriber>)subscriber {
	NSCAssert(NO, @"This method must be overridden by subclasses");
	return nil;
}

- (YSRACDisposable *)subscribeNext:(void (^)(id x))nextBlock {
	NSCParameterAssert(nextBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:nextBlock error:NULL completed:NULL];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeNext:(void (^)(id x))nextBlock completed:(void (^)(void))completedBlock {
	NSCParameterAssert(nextBlock != NULL);
	NSCParameterAssert(completedBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:nextBlock error:NULL completed:completedBlock];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock completed:(void (^)(void))completedBlock {
	NSCParameterAssert(nextBlock != NULL);
	NSCParameterAssert(errorBlock != NULL);
	NSCParameterAssert(completedBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:nextBlock error:errorBlock completed:completedBlock];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeError:(void (^)(NSError *error))errorBlock {
	NSCParameterAssert(errorBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:NULL error:errorBlock completed:NULL];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeCompleted:(void (^)(void))completedBlock {
	NSCParameterAssert(completedBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:NULL error:NULL completed:completedBlock];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeNext:(void (^)(id x))nextBlock error:(void (^)(NSError *error))errorBlock {
	NSCParameterAssert(nextBlock != NULL);
	NSCParameterAssert(errorBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:nextBlock error:errorBlock completed:NULL];
	return [self subscribe:o];
}

- (YSRACDisposable *)subscribeError:(void (^)(NSError *))errorBlock completed:(void (^)(void))completedBlock {
	NSCParameterAssert(completedBlock != NULL);
	NSCParameterAssert(errorBlock != NULL);
	
	YSRACSubscriber *o = [YSRACSubscriber subscriberWithNext:NULL error:errorBlock completed:completedBlock];
	return [self subscribe:o];
}

@end

@implementation YSRACSignal (Debugging)

- (YSRACSignal *)logAll {
	return [[[self logNext] logError] logCompleted];
}

- (YSRACSignal *)logNext {
	return [[self doNext:^(id x) {
		NSLog(@"%@ next: %@", self, x);
	}] setNameWithFormat:@"%@", self.name];
}

- (YSRACSignal *)logError {
	return [[self doError:^(NSError *error) {
		NSLog(@"%@ error: %@", self, error);
	}] setNameWithFormat:@"%@", self.name];
}

- (YSRACSignal *)logCompleted {
	return [[self doCompleted:^{
		NSLog(@"%@ completed", self);
	}] setNameWithFormat:@"%@", self.name];
}

@end

@implementation YSRACSignal (Testing)

static const NSTimeInterval YSRACSignalAsynchronousWaitTimeout = 10;

- (id)asynchronousFirstOrDefault:(id)defaultValue success:(BOOL *)success error:(NSError **)error {
	return [self asynchronousFirstOrDefault:defaultValue success:success error:error timeout:YSRACSignalAsynchronousWaitTimeout];
}

- (id)asynchronousFirstOrDefault:(id)defaultValue success:(BOOL *)success error:(NSError **)error timeout:(NSTimeInterval)timeout {
	NSCAssert([NSThread isMainThread], @"%s should only be used from the main thread", __func__);

	__block id result = defaultValue;
	__block BOOL done = NO;

	// Ensures that we don't pass values across thread boundaries by reference.
	__block NSError *localError;
	__block BOOL localSuccess = YES;

	[[[[self
		take:1]
		timeout:timeout onScheduler:[YSRACScheduler scheduler]]
		deliverOn:YSRACScheduler.mainThreadScheduler]
		subscribeNext:^(id x) {
			result = x;
			done = YES;
		} error:^(NSError *e) {
			if (!done) {
				localSuccess = NO;
				localError = e;
				done = YES;
			}
		} completed:^{
			done = YES;
		}];
	
	do {
		[NSRunLoop.mainRunLoop runMode:NSDefaultRunLoopMode beforeDate:[NSDate dateWithTimeIntervalSinceNow:0.1]];
	} while (!done);

	if (success != NULL) *success = localSuccess;
	if (error != NULL) *error = localError;

	return result;
}

- (BOOL)asynchronouslyWaitUntilCompleted:(NSError **)error timeout:(NSTimeInterval)timeout {
	BOOL success = NO;
	[[self ignoreValues] asynchronousFirstOrDefault:nil success:&success error:error timeout:timeout];
	return success;
}

- (BOOL)asynchronouslyWaitUntilCompleted:(NSError **)error {
	return [self asynchronouslyWaitUntilCompleted:error timeout:YSRACSignalAsynchronousWaitTimeout];
}

@end
