function Rx.flatMap(i,k)assert(type(i)=="\102\117\110\099\116\105\111\110","\066\097\100 \112\114\111\106\101\099\116")return function(Q)assert(Observable.isObservable(Q),"\066\097\100 \111\098\115\101\114\118\097\098\108\101")return Observable.new(function(y)local T=Maid.new()local Y=0 local C=false T:GiveTask(Q:Subscribe(function(...)local Q=...local A=i(...)assert(Observable.isObservable(A),"\066\097\100 \111\098\115\101\114\118\097\098\108\101 \102\114\111\109 \112\114\111\106\101\099\116")Y=Y+1 local j=Maid.new()j:GiveTask(A:Subscribe(function(...)if k then y:Fire(k(Q,...))else y:Fire(...)end end,function(...)y:Fail(...)end,function()j:DoCleaning()Y=Y-1 if Y==0 and C then y:Complete()T:DoCleaning()end end))local H=T:GiveTask(j)j:GiveTask(function()T[H]=nil end)end,function(...)y:Fail(...)T:DoCleaning()end,function()C=true if Y==0 then y:Complete()T:DoCleaning()end end))return T end)end end function Rx.switchMap(i)return Rx.pipe({Rx.map(i),Rx.switchAll()})end function Rx.takeUntil(i)assert(Observable.isObservable(i))return function(k)assert(Observable.isObservable(k),"\066\097\100 \111\098\115\101\114\118\097\098\108\101")return Observable.new(function(Q)local y=Maid.new()local T=false local function Y()y:DoCleaning()T=true end y:GiveTask(i:Subscribe(Y,Y,nil))if T then y:DoCleaning()return nil end y:GiveTask(k:Subscribe(Q:GetFireFailComplete()))return y end)end end
--[=[
	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
