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

#import "YSRACMulticastConnection.h"
#import "YSRACMulticastConnection+Private.h"
#import "YSRACDisposable.h"
#import "YSRACSerialDisposable.h"
#import "YSRACSubject.h"
#import <libkern/OSAtomic.h>

@interface YSRACMulticastConnection () {
	YSRACSubject *_signal;

	// When connecting, a caller should attempt to atomically swap the value of this
	// from `0` to `1`.
	//
	// If the swap is successful the caller is resposible for subscribing `_signal`
	// to `sourceSignal` and storing the returned disposable in `serialDisposable`.
	//
	// If the swap is unsuccessful it means that `_sourceSignal` has already been
	// connected and the caller has no action to take.
	int32_t volatile _hasConnected;
}

@property (nonatomic, readonly, strong) YSRACSignal *sourceSignal;
@property (strong) YSRACSerialDisposable *serialDisposable;
@end

@implementation YSRACMulticastConnection

#pragma mark Lifecycle

- (instancetype)initWithSourceSignal:(YSRACSignal *)source subject:(YSRACSubject *)subject {
	NSCParameterAssert(source != nil);
	NSCParameterAssert(subject != nil);

	self = [super init];

	_sourceSignal = source;
	_serialDisposable = [[YSRACSerialDisposable alloc] init];
	_signal = subject;
	
	return self;
}

#pragma mark Connecting

- (YSRACDisposable *)connect {
	BOOL shouldConnect = OSAtomicCompareAndSwap32Barrier(0, 1, &_hasConnected);

	if (shouldConnect) {
		self.serialDisposable.disposable = [self.sourceSignal subscribe:_signal];
	}

	return self.serialDisposable;
}

- (YSRACSignal *)autoconnect {
	__block volatile int32_t subscriberCount = 0;

	return [[YSRACSignal
		createSignal:^(id<YSRACSubscriber> subscriber) {
			OSAtomicIncrement32Barrier(&subscriberCount);

			YSRACDisposable *subscriptionDisposable = [self.signal subscribe:subscriber];
			YSRACDisposable *connectionDisposable = [self connect];

			return [YSRACDisposable disposableWithBlock:^{
				[subscriptionDisposable dispose];

				if (OSAtomicDecrement32Barrier(&subscriberCount) == 0) {
					[connectionDisposable dispose];
				}
			}];
		}]
		setNameWithFormat:@"[%@] -autoconnect", self.signal.name];
}

@end
