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

#import "YSRACCommand.h"
#import "YSRACEXTScope.h"
#import "NSArray+YSRACSequenceAdditions.h"
#import "NSObject+YSRACDeallocating.h"
#import "NSObject+YSRACDescription.h"
#import "NSObject+YSRACPropertySubscribing.h"
#import "YSRACMulticastConnection.h"
#import "YSRACReplaySubject.h"
#import "YSRACScheduler.h"
#import "YSRACSequence.h"
#import "YSRACSignal+Operations.h"
#import <libkern/OSAtomic.h>

NSErrorDomain const YSRACCommandErrorDomain = @"YSRACCommandErrorDomain";
NSString * const YSRACUnderlyingCommandErrorKey = @"YSRACUnderlyingCommandErrorKey";

@interface YSRACCommand () {
	// Atomic backing variable for `allowsConcurrentExecution`.
	volatile uint32_t _allowsConcurrentExecution;
}

/// A subject that sends added execution signals.
@property (nonatomic, strong, readonly) YSRACSubject *addedExecutionSignalsSubject;

/// A subject that sends the new value of `allowsConcurrentExecution` whenever it changes.
@property (nonatomic, strong, readonly) YSRACSubject *allowsConcurrentExecutionSubject;

// `enabled`, but without a hop to the main thread.
//
// Values from this signal may arrive on any thread.
@property (nonatomic, strong, readonly) YSRACSignal *immediateEnabled;

// The signal block that the receiver was initialized with.
@property (nonatomic, copy, readonly) YSRACSignal * (^signalBlock)(id input);

@end

@implementation YSRACCommand

#pragma mark Properties

- (BOOL)allowsConcurrentExecution {
	return _allowsConcurrentExecution != 0;
}

- (void)setAllowsConcurrentExecution:(BOOL)allowed {
	if (allowed) {
		OSAtomicOr32Barrier(1, &_allowsConcurrentExecution);
	} else {
		OSAtomicAnd32Barrier(0, &_allowsConcurrentExecution);
	}

	[self.allowsConcurrentExecutionSubject sendNext:@(_allowsConcurrentExecution)];
}

#pragma mark Lifecycle

- (instancetype)init {
	NSCAssert(NO, @"Use -initWithSignalBlock: instead");
	return nil;
}

- (instancetype)initWithSignalBlock:(YSRACSignal<id> * (^)(id input))signalBlock {
	return [self initWithEnabled:nil signalBlock:signalBlock];
}

- (void)dealloc {
	[_addedExecutionSignalsSubject sendCompleted];
	[_allowsConcurrentExecutionSubject sendCompleted];
}

- (instancetype)initWithEnabled:(YSRACSignal *)enabledSignal signalBlock:(YSRACSignal<id> * (^)(id input))signalBlock {
	NSCParameterAssert(signalBlock != nil);

	self = [super init];

	_addedExecutionSignalsSubject = [YSRACSubject new];
	_allowsConcurrentExecutionSubject = [YSRACSubject new];
	_signalBlock = [signalBlock copy];

	_executionSignals = [[[self.addedExecutionSignalsSubject
		map:^(YSRACSignal *signal) {
			return [signal catchTo:[YSRACSignal empty]];
		}]
		deliverOn:YSRACScheduler.mainThreadScheduler]
		setNameWithFormat:@"%@ -executionSignals", self];
	
	// `errors` needs to be multicasted so that it picks up all
	// `activeExecutionSignals` that are added.
	//
	// In other words, if someone subscribes to `errors` _after_ an execution
	// has started, it should still receive any error from that execution.
	YSRACMulticastConnection *errorsConnection = [[[self.addedExecutionSignalsSubject
		flattenMap:^(YSRACSignal *signal) {
			return [[signal
				ignoreValues]
				catch:^(NSError *error) {
					return [YSRACSignal return:error];
				}];
		}]
		deliverOn:YSRACScheduler.mainThreadScheduler]
		publish];
	
	_errors = [errorsConnection.signal setNameWithFormat:@"%@ -errors", self];
	[errorsConnection connect];

	YSRACSignal *immediateExecuting = [[[[self.addedExecutionSignalsSubject
		flattenMap:^(YSRACSignal *signal) {
			return [[[signal
				catchTo:[YSRACSignal empty]]
				then:^{
					return [YSRACSignal return:@-1];
				}]
				startWith:@1];
		}]
		scanWithStart:@0 reduce:^(NSNumber *running, NSNumber *next) {
			return @(running.integerValue + next.integerValue);
		}]
		map:^(NSNumber *count) {
			return @(count.integerValue > 0);
		}]
		startWith:@NO];

	_executing = [[[[[immediateExecuting
		deliverOn:YSRACScheduler.mainThreadScheduler]
		// This is useful before the first value arrives on the main thread.
		startWith:@NO]
		distinctUntilChanged]
		replayLast]
		setNameWithFormat:@"%@ -executing", self];
	
	YSRACSignal *moreExecutionsAllowed = [YSRACSignal
		if:[self.allowsConcurrentExecutionSubject startWith:@NO]
		then:[YSRACSignal return:@YES]
		else:[immediateExecuting not]];
	
	if (enabledSignal == nil) {
		enabledSignal = [YSRACSignal return:@YES];
	} else {
		enabledSignal = [enabledSignal startWith:@YES];
	}
	
	_immediateEnabled = [[[[YSRACSignal
		combineLatest:@[ enabledSignal, moreExecutionsAllowed ]]
		and]
		takeUntil:self.ysrac_willDeallocSignal]
		replayLast];
	
	_enabled = [[[[[self.immediateEnabled
		take:1]
		concat:[[self.immediateEnabled skip:1] deliverOn:YSRACScheduler.mainThreadScheduler]]
		distinctUntilChanged]
		replayLast]
		setNameWithFormat:@"%@ -enabled", self];

	return self;
}

#pragma mark Execution

- (YSRACSignal *)execute:(id)input {
	// `immediateEnabled` is guaranteed to send a value upon subscription, so
	// -first is acceptable here.
	BOOL enabled = [[self.immediateEnabled first] boolValue];
	if (!enabled) {
		NSError *error = [NSError errorWithDomain:YSRACCommandErrorDomain code:YSRACCommandErrorNotEnabled userInfo:@{
			NSLocalizedDescriptionKey: NSLocalizedString(@"The command is disabled and cannot be executed", nil),
			YSRACUnderlyingCommandErrorKey: self
		}];

		return [YSRACSignal error:error];
	}

	YSRACSignal *signal = self.signalBlock(input);
	NSCAssert(signal != nil, @"nil signal returned from signal block for value: %@", input);

	// We subscribe to the signal on the main thread so that it occurs _after_
	// -addActiveExecutionSignal: completes below.
	//
	// This means that `executing` and `enabled` will send updated values before
	// the signal actually starts performing work.
	YSRACMulticastConnection *connection = [[signal
		subscribeOn:YSRACScheduler.mainThreadScheduler]
		multicast:[YSRACReplaySubject subject]];
	
	[self.addedExecutionSignalsSubject sendNext:connection.signal];

	[connection connect];
	return [connection.signal setNameWithFormat:@"%@ -execute: %@", self, YSRACDescription(input)];
}

@end
