--[=[
	Sort of equivalent of promise.then()

	This takes a stream of observables

	@param project (value: T) -> Observable<U>
	@param resultSelector ((initialValue: T, outputValue: U) -> U)?
	@return (source: Observable<T>) -> Observable<U>
]=]

function Rx.flatMap(project, resultSelector)
	assert(type(project) == "function", "Bad project")

	return function(source)
		assert(Observable.isObservable(source), "Bad observable")

		return Observable.new(function(sub)
			local maid = Maid.new()

			local pendingCount = 0
			local topComplete = false

			maid:GiveTask(source:Subscribe(
				function(...)
					local outerValue = ...

					local observable = project(...)
					assert(Observable.isObservable(observable), "Bad observable from project")

					pendingCount = pendingCount + 1

					local innerMaid = Maid.new()

					innerMaid:GiveTask(observable:Subscribe(
						function(...)
							-- Merge each inner observable
							if resultSelector then
								sub:Fire(resultSelector(outerValue, ...))
							else
								sub:Fire(...)
							end
						end,
						function(...)
							sub:Fail(...)
						end, -- Emit failure automatically
						function()
							innerMaid:DoCleaning()
							pendingCount = pendingCount - 1
							if pendingCount == 0 and topComplete then
								sub:Complete()
								maid:DoCleaning()
							end
						end))

					local key = maid:GiveTask(innerMaid)

					-- Cleanup
					innerMaid:GiveTask(function()
						maid[key] = nil
					end)
				end,
				function(...)
					sub:Fail(...) -- Also reflect failures up to the top!
					maid:DoCleaning()
				end,
				function()
					topComplete = true
					if pendingCount == 0 then
						sub:Complete()
						maid:DoCleaning()
					end
				end))

			return maid
		end)
	end
end

function Rx.switchMap(project)
	return Rx.pipe({
		Rx.map(project);
		Rx.switchAll();
	})
end

function Rx.takeUntil(notifier)
	assert(Observable.isObservable(notifier))

	return function(source)
		assert(Observable.isObservable(source), "Bad observable")

		return Observable.new(function(sub)
			local maid = Maid.new()
			local cancelled = false

			local function cancel()
				maid:DoCleaning()
				cancelled = true
			end

			-- Any value emitted will cancel (complete without any values allows all values to pass)
			maid:GiveTask(notifier:Subscribe(cancel, cancel, nil))

			-- Cancelled immediately? Oh boy.
			if cancelled then
				maid:DoCleaning()
				return nil
			end

			-- Subscribe!
			maid:GiveTask(source:Subscribe(sub:GetFireFailComplete()))

			return maid
		end)
	end
end
