/** @license MIT License (c) copyright 2010-2015 original author or authors */
/** @author Brian Cavalier */
/** @author John Hann */

var Stream = require('../Stream');
var ChainDisposable = require('../disposable/ChainDisposable');
var EmptyDisposable = require('../disposable/EmptyDisposable');

exports.switch = switchLatest;

/**
 * Given a stream of streams, return a new stream that adopts the behavior
 * of the most recent inner stream.
 * @param {Stream} stream of streams on which to switch
 * @returns {Stream} switching stream
 */
function switchLatest(stream) {
	return new Stream(new Switch(stream.source));
}

function Switch(source) {
	this.source = source;
}

Switch.prototype.run = function(sink, scheduler) {
	var switchSink = new SwitchSink(sink, scheduler);
	return new ChainDisposable(switchSink, this.source.run(switchSink, scheduler));
};

function SwitchSink(sink, scheduler) {
	this.sink = sink;
	this.scheduler = scheduler;
	this.current = null;
	this.ended = false;
}

SwitchSink.prototype.event = function(t, stream) {
	this._disposeCurrent(t); // TODO: capture the result of this dispose
	this.current = new Segment(t, Infinity, this, this.sink);
	this.current.disposable = stream.source.run(this.current, this.scheduler);
};

SwitchSink.prototype.end = function(t, x) {
	this.ended = true;
	this._checkEnd(t, x);
};

SwitchSink.prototype.error = function(t, e) {
	this.ended = true;
	this.sink.error(t, e);
};

SwitchSink.prototype.dispose = function() {
	return this._disposeCurrent(0);
};

SwitchSink.prototype._disposeCurrent = function(t) {
	if(this.current !== null) {
		return this.current._dispose(t);
	}
};

SwitchSink.prototype._disposeInner = function(t, inner) {
	inner._dispose(t); // TODO: capture the result of this dispose
	if(inner === this.current) {
		this.current = null;
	}
};

SwitchSink.prototype._checkEnd = function(t, x) {
	if(this.ended && this.current === null) {
		this.sink.end(t, x);
	}
};

SwitchSink.prototype._endInner = function(t, x, inner) {
	this._disposeInner(t, inner);
	this._checkEnd(t, x);
};

SwitchSink.prototype._errorInner = function(t, e, inner) {
	this._disposeInner(t, inner);
	this.sink.error(t, e);
};

function Segment(min, max, outer, sink) {
	this.min = min;
	this.max = max;
	this.outer = outer;
	this.sink = sink;
	this.disposable = new EmptyDisposable();
}

Segment.prototype.event = function(t, x) {
	if(t < this.max) {
		this.sink.event(Math.max(t, this.min), x);
	}
};

Segment.prototype.end = function(t, x) {
	this.outer._endInner(Math.max(t, this.min), x, this);
};

Segment.prototype.error = function(t, e) {
	this.outer._errorInner(Math.max(t, this.min), e, this);
};

Segment.prototype._dispose = function(t) {
	this.max = t;
	return this.disposable.dispose();
};

/** @license MIT License (c) copyright 2010-2015 original author or authors */
/** @author Brian Cavalier */
/** @author John Hann */

var Stream = require('../Stream');
var MulticastSource = require('../source/MulticastSource');
var until = require('./timeslice').takeUntil;
var concurrentMergeMap = require('./concurrentMergeMap').concurrentMergeMap;

exports.switch = switchLatest;

/**
 * Given a stream of streams, return a new stream that adopts the behavior
 * of the most recent inner stream.
 * @param {Stream} stream of streams on which to switch
 * @returns {Stream} switching stream
 */
function switchLatest(stream) {
	var upstream = new Stream(new MulticastSource(stream.source));

	return concurrentMergeMap(untilNext, 1, upstream);

	function untilNext(s) {
		return until(upstream, s);
	}
}

Simplify switch() by using until()