From: Helge Norberg <helge.norberg@svt.se>
Date: Wed, 1 Jul 2015 12:34:42 +0000 (+0200)
Subject: * Refactored so that frame_consumers are stored in a frame_consumer_registry instance... 
X-Git-Tag: 2.1.0_Beta1~320
X-Git-Url: https://git.sesse.net/?a=commitdiff_plain;ds=inline;h=d936ee9d0aae8ffa4b6c52510a313c00e0f48d5e;p=casparcg

* Refactored so that frame_consumers are stored in a frame_consumer_registry instance instead of being stored in global variables.
---

diff --git a/common/memory.h b/common/memory.h
index 3606c3a8e..5d2293a48 100644
--- a/common/memory.h
+++ b/common/memory.h
@@ -715,6 +715,12 @@ shared_ptr<T> make_shared(P0&& p0, P1&& p1, P2&& p2, P3&& p3, P4&& p4, P5&& p5,
     return shared_ptr<T>(std::make_shared<T>(std::forward<P0>(p0), std::forward<P1>(p1), std::forward<P2>(p2), std::forward<P3>(p3), std::forward<P4>(p4), std::forward<P5>(p5), std::forward<P6>(p6), std::forward<P7>(p7)));
 }
 
+template<typename T, typename P0, typename P1, typename P2, typename P3, typename P4, typename P5, typename P6, typename P7, typename P8>
+shared_ptr<T> make_shared(P0&& p0, P1&& p1, P2&& p2, P3&& p3, P4&& p4, P5&& p5, P6&& p6, P7&& p7, P8&& p8)
+{
+	return shared_ptr<T>(std::make_shared<T>(std::forward<P0>(p0), std::forward<P1>(p1), std::forward<P2>(p2), std::forward<P3>(p3), std::forward<P4>(p4), std::forward<P5>(p5), std::forward<P6>(p6), std::forward<P7>(p7), std::forward<P8>(p8)));
+}
+
 template<typename T>
 shared_ptr<T>::shared_ptr() 
     : p_(make_shared<T>())
diff --git a/core/consumer/frame_consumer.cpp b/core/consumer/frame_consumer.cpp
index b695f5a84..07d916d05 100644
--- a/core/consumer/frame_consumer.cpp
+++ b/core/consumer/frame_consumer.cpp
@@ -36,20 +36,28 @@
 #include <map>
 
 namespace caspar { namespace core {
-		
-std::vector<consumer_factory_t> g_consumer_factories;
-std::map<std::wstring, preconfigured_consumer_factory_t> g_preconfigured_consumer_factories;
 
-void register_consumer_factory(const consumer_factory_t& factory)
+struct frame_consumer_registry::impl
+{
+	std::vector<consumer_factory_t> consumer_factories;
+	std::map<std::wstring, preconfigured_consumer_factory_t> preconfigured_consumer_factories;
+};
+
+frame_consumer_registry::frame_consumer_registry()
+	: impl_(new impl)
+{
+}
+
+void frame_consumer_registry::register_consumer_factory(const consumer_factory_t& factory)
 {
-	g_consumer_factories.push_back(factory);
+	impl_->consumer_factories.push_back(factory);
 }
 
-void register_preconfigured_consumer_factory(
+void frame_consumer_registry::register_preconfigured_consumer_factory(
 		const std::wstring& element_name,
 		const preconfigured_consumer_factory_t& factory)
 {
-	g_preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
+	impl_->preconfigured_consumer_factories.insert(std::make_pair(element_name, factory));
 }
 
 class destroy_consumer_proxy : public frame_consumer
@@ -231,14 +239,15 @@ public:
 	monitor::subject& monitor_output() override								{return consumer_->monitor_output();}										
 };
 
-spl::shared_ptr<core::frame_consumer> create_consumer(
-		const std::vector<std::wstring>& params, interaction_sink* sink)
+spl::shared_ptr<core::frame_consumer> frame_consumer_registry::create_consumer(
+		const std::vector<std::wstring>& params, interaction_sink* sink) const
 {
 	if(params.empty())
 		CASPAR_THROW_EXCEPTION(invalid_argument() << arg_name_info("params") << arg_value_info(""));
 	
 	auto consumer = frame_consumer::empty();
-	std::any_of(g_consumer_factories.begin(), g_consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
+	auto& consumer_factories = impl_->consumer_factories;
+	std::any_of(consumer_factories.begin(), consumer_factories.end(), [&](const consumer_factory_t& factory) -> bool
 		{
 			try
 			{
@@ -261,14 +270,15 @@ spl::shared_ptr<core::frame_consumer> create_consumer(
 			   std::move(consumer)))));
 }
 
-spl::shared_ptr<frame_consumer> create_consumer(
+spl::shared_ptr<frame_consumer> frame_consumer_registry::create_consumer(
 		const std::wstring& element_name,
 		const boost::property_tree::wptree& element,
-		interaction_sink* sink)
+		interaction_sink* sink) const
 {
-	auto found = g_preconfigured_consumer_factories.find(element_name);
+	auto& preconfigured_consumer_factories = impl_->preconfigured_consumer_factories;
+	auto found = preconfigured_consumer_factories.find(element_name);
 
-	if (found == g_preconfigured_consumer_factories.end())
+	if (found == preconfigured_consumer_factories.end())
 		CASPAR_THROW_EXCEPTION(file_not_found()
 			<< msg_info(L"No consumer factory registered for element name " + element_name));
 
diff --git a/core/consumer/frame_consumer.h b/core/consumer/frame_consumer.h
index 3a0f7ec62..fbda2c460 100644
--- a/core/consumer/frame_consumer.h
+++ b/core/consumer/frame_consumer.h
@@ -77,16 +77,24 @@ typedef std::function<spl::shared_ptr<frame_consumer>(
 		const boost::property_tree::wptree& element,
 		interaction_sink* sink)> preconfigured_consumer_factory_t;
 
-void register_consumer_factory(const consumer_factory_t& factory);
-void register_preconfigured_consumer_factory(
-		const std::wstring& element_name,
-		const preconfigured_consumer_factory_t& factory);
-spl::shared_ptr<frame_consumer> create_consumer(
-		const std::vector<std::wstring>& params,
-		interaction_sink* sink);
-spl::shared_ptr<frame_consumer> create_consumer(
-		const std::wstring& element_name,
-		const boost::property_tree::wptree& element,
-		interaction_sink* sink);
+class frame_consumer_registry : boost::noncopyable
+{
+public:
+	frame_consumer_registry();
+	void register_consumer_factory(const consumer_factory_t& factory);
+	void register_preconfigured_consumer_factory(
+			const std::wstring& element_name,
+			const preconfigured_consumer_factory_t& factory);
+	spl::shared_ptr<frame_consumer> create_consumer(
+			const std::vector<std::wstring>& params,
+			interaction_sink* sink) const;
+	spl::shared_ptr<frame_consumer> create_consumer(
+			const std::wstring& element_name,
+			const boost::property_tree::wptree& element,
+			interaction_sink* sink) const;
+private:
+	struct impl;
+	spl::shared_ptr<impl> impl_;
+};
 
 }}
diff --git a/core/module_dependencies.h b/core/module_dependencies.h
index 1357e7c99..3fe9cfe0b 100644
--- a/core/module_dependencies.h
+++ b/core/module_dependencies.h
@@ -27,6 +27,7 @@
 #include "producer/cg_proxy.h"
 #include "producer/media_info/media_info_repository.h"
 #include "producer/frame_producer.h"
+#include "consumer/frame_consumer.h"
 
 namespace caspar { namespace core {
 
@@ -36,16 +37,19 @@ struct module_dependencies
 	const spl::shared_ptr<cg_producer_registry>				cg_registry;
 	const spl::shared_ptr<media_info_repository>			media_info_repo;
 	const spl::shared_ptr<frame_producer_registry>			producer_registry;
+	const spl::shared_ptr<frame_consumer_registry>			consumer_registry;
 
 	module_dependencies(
 			spl::shared_ptr<system_info_provider_repository> system_info_provider_repo,
 			spl::shared_ptr<cg_producer_registry> cg_registry,
 			spl::shared_ptr<media_info_repository> media_info_repo,
-			spl::shared_ptr<frame_producer_registry> producer_registry)
+			spl::shared_ptr<frame_producer_registry> producer_registry,
+			spl::shared_ptr<frame_consumer_registry> consumer_registry)
 		: system_info_provider_repo(std::move(system_info_provider_repo))
 		, cg_registry(std::move(cg_registry))
 		, media_info_repo(std::move(media_info_repo))
 		, producer_registry(std::move(producer_registry))
+		, consumer_registry(std::move(consumer_registry))
 	{
 	}
 };
diff --git a/modules/bluefish/bluefish.cpp b/modules/bluefish/bluefish.cpp
index 34a83a33f..d2ecabf83 100644
--- a/modules/bluefish/bluefish.cpp
+++ b/modules/bluefish/bluefish.cpp
@@ -84,8 +84,8 @@ void init(core::module_dependencies dependencies)
 	}
 	catch(...){}
 
-	core::register_consumer_factory(create_consumer);
-	core::register_preconfigured_consumer_factory(L"bluefish", create_preconfigured_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"bluefish", create_preconfigured_consumer);
 	dependencies.system_info_provider_repo->register_system_info_provider([](boost::property_tree::wptree& info)
 	{
 		info.add(L"system.bluefish.version", version());
diff --git a/modules/decklink/decklink.cpp b/modules/decklink/decklink.cpp
index 71454d42c..1f0424ebc 100644
--- a/modules/decklink/decklink.cpp
+++ b/modules/decklink/decklink.cpp
@@ -82,8 +82,8 @@ std::vector<std::wstring> device_list()
 
 void init(core::module_dependencies dependencies)
 {
-	core::register_consumer_factory(create_consumer);
-	core::register_preconfigured_consumer_factory(L"decklink", create_preconfigured_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"decklink", create_preconfigured_consumer);
 	dependencies.producer_registry->register_producer_factory(create_producer);
 	dependencies.system_info_provider_repo->register_system_info_provider([](boost::property_tree::wptree& info)
 	{
diff --git a/modules/ffmpeg/ffmpeg.cpp b/modules/ffmpeg/ffmpeg.cpp
index 66ebb1c9c..ad0c2e9ba 100644
--- a/modules/ffmpeg/ffmpeg.cpp
+++ b/modules/ffmpeg/ffmpeg.cpp
@@ -247,10 +247,10 @@ void init(core::module_dependencies dependencies)
     avformat_network_init();
     avcodec_register_all();
 	
-	core::register_consumer_factory(create_consumer);
-	core::register_consumer_factory(create_streaming_consumer);
-	core::register_preconfigured_consumer_factory(L"file", create_preconfigured_consumer);
-	core::register_preconfigured_consumer_factory(L"stream", create_preconfigured_streaming_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_streaming_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"file", create_preconfigured_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"stream", create_preconfigured_streaming_consumer);
 	dependencies.producer_registry->register_producer_factory(create_producer);
 	
 	dependencies.media_info_repo->register_extractor(
diff --git a/modules/image/image.cpp b/modules/image/image.cpp
index f68dab212..04adae102 100644
--- a/modules/image/image.cpp
+++ b/modules/image/image.cpp
@@ -50,7 +50,7 @@ void init(core::module_dependencies dependencies)
 	dependencies.producer_registry->register_producer_factory(create_scroll_producer);
 	dependencies.producer_registry->register_producer_factory(create_producer);
 	dependencies.producer_registry->register_thumbnail_producer_factory(create_thumbnail_producer);
-	core::register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
 	dependencies.media_info_repo->register_extractor([](const std::wstring& file, const std::wstring& extension, core::media_info& info)
 	{
 		if (extension == L".TGA"
diff --git a/modules/newtek/newtek.cpp b/modules/newtek/newtek.cpp
index 839530cce..15ffe1801 100644
--- a/modules/newtek/newtek.cpp
+++ b/modules/newtek/newtek.cpp
@@ -33,14 +33,14 @@ void init(core::module_dependencies dependencies)
 	try
 	{
 		if (airsend::is_available())
-			core::register_consumer_factory(create_ivga_consumer);
+			dependencies.consumer_registry->register_consumer_factory(create_ivga_consumer);
 		dependencies.system_info_provider_repo->register_system_info_provider([](boost::property_tree::wptree& info)
 		{
 			info.add(L"system.newtek-ivga.version", airsend::is_available()
 					? L"available"
 					: L"unavailable (" + airsend::dll_name() + L")");
 		});
-		core::register_preconfigured_consumer_factory(L"newtek-ivga", create_preconfigured_ivga_consumer);
+		dependencies.consumer_registry->register_preconfigured_consumer_factory(L"newtek-ivga", create_preconfigured_ivga_consumer);
 	}
 	catch(...){}
 }
diff --git a/modules/oal/oal.cpp b/modules/oal/oal.cpp
index 77321f9d2..ac3d79d1d 100644
--- a/modules/oal/oal.cpp
+++ b/modules/oal/oal.cpp
@@ -29,8 +29,8 @@ namespace caspar { namespace oal {
 
 void init(core::module_dependencies dependencies)
 {
-	core::register_consumer_factory(create_consumer);
-	core::register_preconfigured_consumer_factory(L"system-audio", create_preconfigured_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"system-audio", create_preconfigured_consumer);
 }
 
 }}
diff --git a/modules/screen/screen.cpp b/modules/screen/screen.cpp
index 15a7ca12a..ef670950a 100644
--- a/modules/screen/screen.cpp
+++ b/modules/screen/screen.cpp
@@ -29,8 +29,8 @@ namespace caspar { namespace screen {
 
 void init(core::module_dependencies dependencies)
 {
-	core::register_consumer_factory(create_consumer);
-	core::register_preconfigured_consumer_factory(L"screen", create_preconfigured_consumer);
+	dependencies.consumer_registry->register_consumer_factory(create_consumer);
+	dependencies.consumer_registry->register_preconfigured_consumer_factory(L"screen", create_preconfigured_consumer);
 }
 
 }}
diff --git a/protocol/amcp/AMCPCommand.h b/protocol/amcp/AMCPCommand.h
index 7cb9553cf..33cc8cf98 100644
--- a/protocol/amcp/AMCPCommand.h
+++ b/protocol/amcp/AMCPCommand.h
@@ -44,6 +44,7 @@ namespace amcp {
 		spl::shared_ptr<core::system_info_provider_repository>	system_info_repo;
 		std::shared_ptr<core::thumbnail_generator>				thumb_gen;
 		spl::shared_ptr<const core::frame_producer_registry>	producer_registry;
+		spl::shared_ptr<const core::frame_consumer_registry>	consumer_registry;
 		std::promise<bool>&										shutdown_server_now;
 		std::vector<std::wstring>								parameters;
 
@@ -61,6 +62,7 @@ namespace amcp {
 				spl::shared_ptr<core::system_info_provider_repository> system_info_repo,
 				std::shared_ptr<core::thumbnail_generator> thumb_gen,
 				spl::shared_ptr<const core::frame_producer_registry> producer_registry,
+				spl::shared_ptr<const core::frame_consumer_registry> consumer_registry,
 				std::promise<bool>& shutdown_server_now)
 			: client(std::move(client))
 			, channel(channel)
@@ -73,6 +75,7 @@ namespace amcp {
 			, system_info_repo(std::move(system_info_repo))
 			, thumb_gen(std::move(thumb_gen))
 			, producer_registry(std::move(producer_registry))
+			, consumer_registry(std::move(consumer_registry))
 			, shutdown_server_now(shutdown_server_now)
 		{
 		}
diff --git a/protocol/amcp/AMCPCommandsImpl.cpp b/protocol/amcp/AMCPCommandsImpl.cpp
index 5a187c9d7..0828e8754 100644
--- a/protocol/amcp/AMCPCommandsImpl.cpp
+++ b/protocol/amcp/AMCPCommandsImpl.cpp
@@ -620,7 +620,7 @@ std::wstring add_command(command_context& ctx)
 	core::diagnostics::scoped_call_context save;
 	core::diagnostics::call_context::for_thread().video_channel = ctx.channel_index + 1;
 
-	auto consumer = create_consumer(ctx.parameters, &ctx.channel.channel->stage());
+	auto consumer = ctx.consumer_registry->create_consumer(ctx.parameters, &ctx.channel.channel->stage());
 	ctx.channel.channel->output().add(ctx.layer_index(consumer->index()), consumer);
 
 	return L"202 ADD OK\r\n";
@@ -653,7 +653,7 @@ std::wstring remove_command(command_context& ctx)
 				ctx.client->address(),
 				ctx.parameters);
 
-		index = create_consumer(ctx.parameters, &ctx.channel.channel->stage())->index();
+		index = ctx.consumer_registry->create_consumer(ctx.parameters, &ctx.channel.channel->stage())->index();
 	}
 
 	ctx.channel.channel->output().remove(index);
@@ -674,7 +674,7 @@ void print_describer(core::help_sink& sink, const core::help_repository& repo)
 
 std::wstring print_command(command_context& ctx)
 {
-	ctx.channel.channel->output().add(create_consumer({ L"IMAGE" }, &ctx.channel.channel->stage()));
+	ctx.channel.channel->output().add(ctx.consumer_registry->create_consumer({ L"IMAGE" }, &ctx.channel.channel->stage()));
 
 	return L"202 PRINT OK\r\n";
 }
@@ -1982,7 +1982,7 @@ std::wstring channel_grid_command(command_context& ctx)
 	params.push_back(L"0");
 	params.push_back(L"NAME");
 	params.push_back(L"Channel Grid Window");
-	auto screen = create_consumer(params, &self.channel->stage());
+	auto screen = ctx.consumer_registry->create_consumer(params, &self.channel->stage());
 
 	self.channel->output().add(screen);
 
diff --git a/protocol/amcp/amcp_command_repository.cpp b/protocol/amcp/amcp_command_repository.cpp
index e1612eb14..b30a0d84b 100644
--- a/protocol/amcp/amcp_command_repository.cpp
+++ b/protocol/amcp/amcp_command_repository.cpp
@@ -70,6 +70,7 @@ struct amcp_command_repository::impl
 	spl::shared_ptr<core::cg_producer_registry>					cg_registry;
 	spl::shared_ptr<core::help_repository>						help_repo;
 	spl::shared_ptr<const core::frame_producer_registry>		producer_registry;
+	spl::shared_ptr<const core::frame_consumer_registry>		consumer_registry;
 	std::promise<bool>&											shutdown_server_now;
 
 	std::map<std::wstring, std::pair<amcp_command_func, int>>	commands;
@@ -83,6 +84,7 @@ struct amcp_command_repository::impl
 			const spl::shared_ptr<core::cg_producer_registry>& cg_registry,
 			const spl::shared_ptr<core::help_repository>& help_repo,
 			const spl::shared_ptr<const core::frame_producer_registry>& producer_registry,
+			const spl::shared_ptr<const core::frame_consumer_registry>& consumer_registry,
 			std::promise<bool>& shutdown_server_now)
 		: thumb_gen(thumb_gen)
 		, media_info_repo(media_info_repo)
@@ -90,6 +92,7 @@ struct amcp_command_repository::impl
 		, cg_registry(cg_registry)
 		, help_repo(help_repo)
 		, producer_registry(producer_registry)
+		, consumer_registry(consumer_registry)
 		, shutdown_server_now(shutdown_server_now)
 	{
 		int index = 0;
@@ -110,8 +113,18 @@ amcp_command_repository::amcp_command_repository(
 		const spl::shared_ptr<core::cg_producer_registry>& cg_registry,
 		const spl::shared_ptr<core::help_repository>& help_repo,
 		const spl::shared_ptr<const core::frame_producer_registry>& producer_registry,
+		const spl::shared_ptr<const core::frame_consumer_registry>& consumer_registry,
 		std::promise<bool>& shutdown_server_now)
-	: impl_(new impl(channels, thumb_gen, media_info_repo, system_info_provider_repo, cg_registry, help_repo, producer_registry, shutdown_server_now))
+		: impl_(new impl(
+				channels,
+				thumb_gen,
+				media_info_repo,
+				system_info_provider_repo,
+				cg_registry,
+				help_repo,
+				producer_registry,
+				consumer_registry,
+				shutdown_server_now))
 {
 }
 
@@ -131,6 +144,7 @@ AMCPCommand::ptr_type amcp_command_repository::create_command(const std::wstring
 			self.system_info_provider_repo,
 			self.thumb_gen,
 			self.producer_registry,
+			self.consumer_registry,
 			self.shutdown_server_now);
 
 	auto command = find_command(self.commands, s, ctx, tokens);
@@ -169,6 +183,7 @@ AMCPCommand::ptr_type amcp_command_repository::create_channel_command(
 			self.system_info_provider_repo,
 			self.thumb_gen,
 			self.producer_registry,
+			self.consumer_registry,
 			self.shutdown_server_now);
 
 	auto command = find_command(self.channel_commands, s, ctx, tokens);
diff --git a/protocol/amcp/amcp_command_repository.h b/protocol/amcp/amcp_command_repository.h
index 0c208b2cf..c92d6ef29 100644
--- a/protocol/amcp/amcp_command_repository.h
+++ b/protocol/amcp/amcp_command_repository.h
@@ -44,6 +44,7 @@ public:
 			const spl::shared_ptr<core::cg_producer_registry>& cg_registry,
 			const spl::shared_ptr<core::help_repository>& help_repo,
 			const spl::shared_ptr<const core::frame_producer_registry>& producer_registry,
+			const spl::shared_ptr<const core::frame_consumer_registry>& consumer_registry,
 			std::promise<bool>& shutdown_server_now);
 
 	AMCPCommand::ptr_type create_command(const std::wstring& s, IO::ClientInfoPtr client, std::list<std::wstring>& tokens) const;
diff --git a/shell/generate_docs.cpp b/shell/generate_docs.cpp
index 57542cf01..78fce5b03 100644
--- a/shell/generate_docs.cpp
+++ b/shell/generate_docs.cpp
@@ -180,6 +180,7 @@ int main(int argc, char** argv)
 	auto media_info_repo = core::create_in_memory_media_info_repository();
 	spl::shared_ptr<core::help_repository> help_repo;
 	spl::shared_ptr<core::frame_producer_registry> producer_registry;
+	spl::shared_ptr<core::frame_consumer_registry> consumer_registry;
 	std::promise<bool> shutdown_server_now;
 	protocol::amcp::amcp_command_repository repo(
 			{ },
@@ -189,11 +190,12 @@ int main(int argc, char** argv)
 			cg_registry,
 			help_repo,
 			producer_registry,
+			consumer_registry,
 			shutdown_server_now);
 
 	protocol::amcp::register_commands(repo);
 
-	core::module_dependencies dependencies(system_info_provider_repo, cg_registry, media_info_repo, producer_registry);
+	core::module_dependencies dependencies(system_info_provider_repo, cg_registry, media_info_repo, producer_registry, consumer_registry);
 	initialize_modules(dependencies);
 
 	generate_amcp_commands_help(*help_repo);
diff --git a/shell/server.cpp b/shell/server.cpp
index 6dd2a4c67..9033cd873 100644
--- a/shell/server.cpp
+++ b/shell/server.cpp
@@ -96,6 +96,7 @@ struct server::impl : boost::noncopyable
 	spl::shared_ptr<system_info_provider_repository>	system_info_provider_repo_;
 	spl::shared_ptr<core::cg_producer_registry>			cg_registry_;
 	spl::shared_ptr<core::frame_producer_registry>		producer_registry_;
+	spl::shared_ptr<core::frame_consumer_registry>		consumer_registry_;
 	tbb::atomic<bool>									running_;
 	std::shared_ptr<thumbnail_generator>				thumbnail_generator_;
 	std::promise<bool>&									shutdown_server_now_;
@@ -114,7 +115,8 @@ struct server::impl : boost::noncopyable
 				system_info_provider_repo_,
 				cg_registry_,
 				media_info_repo_,
-				producer_registry_);
+				producer_registry_,
+				consumer_registry_);
 
 		initialize_modules(dependencies);
 		core::text::init(dependencies);
@@ -182,7 +184,7 @@ struct server::impl : boost::noncopyable
 					auto name = xml_consumer.first;
 					
 					if (name != L"<xmlcomment>")
-						channel->output().add(create_consumer(name, xml_consumer.second, &channel->stage()));
+						channel->output().add(consumer_registry_->create_consumer(name, xml_consumer.second, &channel->stage()));
 				}
 				catch(...)
 				{
@@ -278,6 +280,7 @@ struct server::impl : boost::noncopyable
 				cg_registry_,
 				help_repo_,
 				producer_registry_,
+				consumer_registry_,
 				shutdown_server_now_);
 		amcp::register_commands(*amcp_command_repo_);