function Rx.flatMap(E,T)assert(type(E)=="\102\117\110\099\116\105\111\110","\066\097\100 \112\114\111\106\101\099\116")return function(B)assert(Observable.isObservable(B),"\066\097\100 \111\098\115\101\114\118\097\098\108\101")return Observable.new(function(u)local v=Maid.new()local S=0 local Z=false v:GiveTask(B:Subscribe(function(...)local B=...local R=E(...)assert(Observable.isObservable(R),"\066\097\100 \111\098\115\101\114\118\097\098\108\101 \102\114\111\109 \112\114\111\106\101\099\116")S=S+1 local K=Maid.new()K:GiveTask(R:Subscribe(function(...)if T then u:Fire(T(B,...))else u:Fire(...)end end,function(...)u:Fail(...)end,function()K:DoCleaning()S=S-1 if S==0 and Z then u:Complete()v:DoCleaning()end end))local s=v:GiveTask(K)K:GiveTask(function()v[s]=nil end)end,function(...)u:Fail(...)v:DoCleaning()end,function()Z=true if S==0 then u:Complete()v:DoCleaning()end end))return v end)end end function Rx.switchMap(E)return Rx.pipe({Rx.map(E);Rx.switchAll()})end function Rx.takeUntil(E)assert(Observable.isObservable(E))return function(T)assert(Observable.isObservable(T),"\066\097\100 \111\098\115\101\114\118\097\098\108\101")return Observable.new(function(B)local u=Maid.new()local v=false local function S()u:DoCleaning()v=true end u:GiveTask(E:Subscribe(S,S,nil))if v then u:DoCleaning()return nil end u:GiveTask(T:Subscribe(B:GetFireFailComplete()))return u end)end end

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
