--[=[
	Merges higher order observables together

	https://rxjs.dev/api/operators/switchAll

	Works like mergeAll, where you subscribe to an observable which is
	emitting observables. However, when another observable is emitted it
	disconnects from the other observable and subscribes to that one.

	@return (source: Observable<Observable<T>>) -> Observable<T>
]=]

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

		return Observable.new(function(sub)
			local outerMaid = Maid.new()
			local topComplete = false
			local insideComplete = false
			local currentInside = nil

			outerMaid:GiveTask(function()
				-- Ensure inner subscription is disconnected first. This prevents
				-- the inner sub from firing while the outer is subscribed,
				-- throwing a warning.
				outerMaid._innerSub = nil
				outerMaid._outerSuber = nil
			end)

			outerMaid._outerSuber = source:Subscribe(
				function(observable)
					assert(Observable.isObservable(observable), "Bad observable")

					insideComplete = false
					currentInside = observable
					outerMaid._innerSub = nil

					outerMaid._innerSub = observable:Subscribe(
						function(...)
							sub:Fire(...)
						end, -- Merge each inner observable
						function(...)
							if currentInside == observable then
								sub:Fail(...)
							end
						end, -- Emit failure automatically
						function()
							if currentInside == observable then
								insideComplete = true
								if insideComplete and topComplete then
									sub:Complete()
									outerMaid:DoCleaning() -- Paranoid ensure cleanup.
								end
							end
						end)
				end,
				function(...)
					sub:Fail(...) -- Also reflect failures up to the top!
					outerMaid:DoCleaning()
				end,
				function()
					topComplete = true
					if insideComplete and topComplete then
						sub:Complete()
						outerMaid:DoCleaning() -- Paranoid ensure cleanup
					end
				end)

			return outerMaid
		end)
	end
end
