--[=[
	Unpacks the brio, and then repacks it. Ignored items
	still invalidate the previous brio

	@since 3.6.0
	@param predicate (T) -> boolean
	@return (source: Observable<Brio<T>>) -> Observable<Brio<T>>
]=]

function RxBrioUtils.where(predicate)
	assert(type(predicate) == "function", "Bad predicate")
	return function(source)
		return Observable.new(function(sub)
			local maid = Maid.new()

			maid:GiveTask(source:Subscribe(function(brio)
				assert(Brio.isBrio(brio), "Not a brio")
				if brio:IsDead() then
					return
				end

				if predicate(brio:GetValue()) then
					sub:Fire(brio)
				end
			end, sub:GetFailComplete()))

			return maid
		end)
	end
end
