--[=[
	One of the most useful functions this combines the latest values of
	observables at each chance!

	```lua
	Rx.combineLatest({
		child = Rx.fromSignal(Workspace.ChildAdded);
		lastChildRemoved = Rx.fromSignal(Workspace.ChildRemoved);
		value = 5;

	}):Subscribe(function(data)
		print(data.child) --> last child
		print(data.lastChildRemoved) --> other value
		print(data.value) --> 5
	end)

	```

	:::tip
	Note that the resulting observable will not emit until all input
	observables are emitted.
	:::

	@param observables { [TKey]: Observable<TEmitted> | TEmitted }
	@return Observable<{ [TKey]: TEmitted }>
]=]

function Rx.combineLatest(observables)
	assert(type(observables) == "table", "Bad observables")

	return Observable.new(function(sub)
		local pending = 0

		local latest = {}
		for key, value in pairs(observables) do
			if Observable.isObservable(value) then
				pending = pending + 1
				latest[key] = UNSET_VALUE
			else
				latest[key] = value
			end
		end

		if pending == 0 then
			sub:Fire(latest)
			sub:Complete()
			return
		end

		local maid = Maid.new()

		local function fireIfAllSet()
			for _, value in pairs(latest) do
				if value == UNSET_VALUE then
					return
				end
			end

			sub:Fire(Table.copy(latest))
		end

		for key, observer in pairs(observables) do
			if Observable.isObservable(observer) then
				maid:GiveTask(observer:Subscribe(
					function(value)
						latest[key] = value
						fireIfAllSet()
					end,
					function(...)
						pending = pending - 1
						sub:Fail(...)
					end,
					function()
						pending = pending - 1
						if pending == 0 then
							sub:Complete()
						end
					end))
			end
		end

		return maid
	end)
end
