function Rx.pipe(p)assert(type(p)=="\116\097\098\108\101","\066\097\100 \116\114\097\110\115\102\111\114\109\101\114\115")for p,X in pairs(p)do if type(X)~="\102\117\110\099\116\105\111\110"then error(("\091\082\120\046\112\105\112\101\093 \066\097\100 \112\105\112\101 \118\097\108\117\101 \111\102 \116\121\112\101 \037\113 \097\116 \105\110\100\101\120 \037\113\044 \101\120\112\101\099\116\101\100 \102\117\110\099\116\105\111\110"):format(type(X),tostring(p)))end end return function(X)assert(X,"\066\097\100 \115\111\117\114\099\101")local S=X for p,X in pairs(p)do S=X(S)if not(type(S)=="\116\097\098\108\101"and S.ClassName=="\079\098\115\101\114\118\097\098\108\101")then error(("\091\082\120\046\112\105\112\101\093 - \070\097\105\108\101\100 \116\111 \116\114\097\110\115\102\111\114\109 \037\113 \105\110 \112\105\112\101\044 \109\097\100\101 \037\113 \040\037\115\041"):format(tostring(p),tostring(S),tostring(type(S)=="\116\097\098\108\101"and S.ClassName or"")))end end return S end end

function Rx.pipe(transformers)
	assert(type(transformers) == "table", "Bad transformers")
	for index, transformer in pairs(transformers) do
		if type(transformer) ~= "function" then
			error(("[Rx.pipe] Bad pipe value of type %q at index %q, expected function")
				:format(type(transformer), tostring(index)))
		end
	end

	return function(source)
		assert(source, "Bad source")

		local current = source
		for key, transformer in pairs(transformers) do
			current = transformer(current)

			if not (type(current) == "table" and current.ClassName == "Observable") then
				error(("[Rx.pipe] - Failed to transform %q in pipe, made %q (%s)")
					:format(tostring(key), tostring(current), tostring(type(current) == "table" and current.ClassName or "")))
			end
		end

		return current
	end
end
