safe_ptr<basic_frame> receive()
	{		
		try
		{
			if(is_paused_)
				return disable_audio(foreground_->last_frame());
		
			auto frame = receive_and_follow(foreground_, frame_producer::NO_HINT);
			if(frame == core::basic_frame::late())
				return foreground_->last_frame();

			auto frames_left = foreground_->nb_frames() - (++frame_number_) - auto_play_delta_;
			if(auto_play_delta_ > -1 && frames_left < 1)
			{
				play();
				return receive();
			}
				
			return frame;
		}
		catch(...)
		{
			CASPAR_LOG_CURRENT_EXCEPTION();
			stop();
			return core::basic_frame::empty();
		}
	}

	safe_ptr<basic_frame> receive(int hints)
	{		
		try
		{
			if(is_paused_)
				return disable_audio(foreground_->last_frame());
		
			auto frame = receive_and_follow(foreground_, hints);
			if(frame == core::basic_frame::late())
				return foreground_->last_frame();

			auto frames_left = static_cast<int64_t>(foreground_->nb_frames()) - static_cast<int64_t>(++frame_number_) - static_cast<int64_t>(auto_play_delta_);
			if(auto_play_delta_ > -1 && frames_left < 1)
			{
				play();
				return receive(hints);
			}
				
			return frame;
		}
		catch(...)
		{
			CASPAR_LOG_CURRENT_EXCEPTION();
			stop();
			return core::basic_frame::empty();
		}
	}

	explicit channel_producer(const safe_ptr<frame_factory>& frame_factory, const safe_ptr<video_channel>& channel) 
		: frame_factory_(frame_factory)
		, consumer_(make_safe<channel_consumer>())
		, last_frame_(basic_frame::empty())
		, frame_number_(0)
	{
		channel->output()->add(consumer_);
		consumer_->block_until_first_frame_available();
		CASPAR_LOG(info) << print() << L" Initialized";
	}

	safe_ptr<core::basic_frame> get_frame(int hints)
	{
		if(exception_ != nullptr)
			std::rethrow_exception(exception_);

		hints_ = hints;

		safe_ptr<core::basic_frame> frame = core::basic_frame::late();
		if(!frame_buffer_.try_pop(frame))
			graph_->set_tag("late-frame");
		graph_->set_value("output-buffer", static_cast<float>(frame_buffer_.size())/static_cast<float>(frame_buffer_.capacity()));	
		return frame;
	}

	explicit separated_producer(const safe_ptr<frame_producer>& fill, const safe_ptr<frame_producer>& key) 
		: monitor_subject_("")
		, key_monitor_subject_("/keyer")
		, fill_producer_(fill)
		, key_producer_(key)
		, fill_(core::basic_frame::late())
		, key_(core::basic_frame::late())
		, last_frame_(core::basic_frame::empty())
	{
		key_monitor_subject_.link_target(&monitor_subject_);

		key_producer_->monitor_output().link_target(&key_monitor_subject_);
		fill_producer_->monitor_output().link_target(&monitor_subject_);
	}

	std::shared_ptr<AVFrame> decode(safe_ptr<AVPacket> pkt)
	{
		std::shared_ptr<AVFrame> decoded_frame(avcodec_alloc_frame(), av_free);

		int frame_finished = 0;
		THROW_ON_ERROR2(avcodec_decode_video2(codec_context_.get(), decoded_frame.get(), &frame_finished, pkt.get()), "[video_decoder]");
		
		// If a decoder consumes less then the whole packet then something is wrong
		// that might be just harmless padding at the end, or a problem with the
		// AVParser or demuxer which puted more then one frame in a AVPacket.

		if(frame_finished == 0)	
			return nullptr;

		is_progressive_ = !decoded_frame->interlaced_frame;

		if(decoded_frame->repeat_pict > 0)
			CASPAR_LOG(warning) << "[video_decoder] Field repeat_pict not implemented.";
		
		++file_frame_number_;

		// This ties the life of the decoded_frame to the packet that it came from. For the
		// current version of ffmpeg (0.8 or c17808c) the RAW_VIDEO codec returns frame data
		// owned by the packet.
		return std::shared_ptr<AVFrame>(decoded_frame.get(), [decoded_frame, pkt](AVFrame*){});
	}

	virtual safe_ptr<basic_frame> create_thumbnail_frame() override
	{
		auto fill_frame = fill_producer_->create_thumbnail_frame();
		auto key_frame = key_producer_->create_thumbnail_frame();

		if (fill_frame == basic_frame::empty() || key_frame == basic_frame::empty())
			return basic_frame::empty();

		if (fill_frame == basic_frame::eof() || key_frame == basic_frame::eof())
			return basic_frame::eof();

		if (fill_frame == basic_frame::late() || key_frame == basic_frame::late())
			return basic_frame::late();

		return basic_frame::fill_and_key(fill_frame, key_frame);
	}

	decklink_producer(const core::video_format_desc& format_desc, size_t device_index, const safe_ptr<core::frame_factory>& frame_factory, const std::wstring& filter)
		: decklink_(get_device(device_index))
		, input_(decklink_)
		, attributes_(decklink_)
		, model_name_(get_model_name(decklink_))
		, device_index_(device_index)
		, filter_(filter)
		, format_desc_(format_desc)
		, audio_cadence_(format_desc.audio_cadence)
		, muxer_(format_desc.fps, frame_factory, filter)
		, sync_buffer_(format_desc.audio_cadence.size())
		, frame_factory_(frame_factory)
	{		
		hints_ = 0;
		frame_buffer_.set_capacity(2);
		
		graph_->set_color("tick-time", diagnostics::color(0.0f, 0.6f, 0.9f));	
		graph_->set_color("late-frame", diagnostics::color(0.6f, 0.3f, 0.3f));
		graph_->set_color("frame-time", diagnostics::color(1.0f, 0.0f, 0.0f));
		graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));
		graph_->set_color("output-buffer", diagnostics::color(0.0f, 1.0f, 0.0f));
		graph_->set_text(print());
		diagnostics::register_graph(graph_);
		
		auto display_mode = get_display_mode(input_, format_desc_.format, bmdFormat8BitYUV, bmdVideoInputFlagDefault);
				
		// NOTE: bmdFormat8BitARGB is currently not supported by any decklink card. (2011-05-08)
		if(FAILED(input_->EnableVideoInput(display_mode, bmdFormat8BitYUV, bmdVideoInputFlagDefault))) 
			BOOST_THROW_EXCEPTION(caspar_exception() 
									<< msg_info(narrow(print()) + " Could not enable video input.")
									<< boost::errinfo_api_function("EnableVideoInput"));

		if(FAILED(input_->EnableAudioInput(bmdAudioSampleRate48kHz, bmdAudioSampleType32bitInteger, format_desc_.audio_channels))) 
			BOOST_THROW_EXCEPTION(caspar_exception() 
									<< msg_info(narrow(print()) + " Could not enable audio input.")
									<< boost::errinfo_api_function("EnableAudioInput"));
			
		if (FAILED(input_->SetCallback(this)) != S_OK)
			BOOST_THROW_EXCEPTION(caspar_exception() 
									<< msg_info(narrow(print()) + " Failed to set input callback.")
									<< boost::errinfo_api_function("SetCallback"));
			
		if(FAILED(input_->StartStreams()))
			BOOST_THROW_EXCEPTION(caspar_exception() 
									<< msg_info(narrow(print()) + " Failed to start input stream.")
									<< boost::errinfo_api_function("StartStreams"));
	}

	virtual safe_ptr<basic_frame> receive(int) override
	{
		auto format_desc = consumer_->get_video_format_desc();

		if(frame_buffer_.size() > 0)
		{
			auto frame = frame_buffer_.front();
			frame_buffer_.pop();
			return last_frame_ = frame;
		}
		
		auto read_frame = consumer_->receive();
		if(!read_frame || read_frame->image_data().empty())
			return basic_frame::late();		

		frame_number_++;
		
		core::pixel_format_desc desc;
		bool double_speed	= std::abs(frame_factory_->get_video_format_desc().fps / 2.0 - format_desc.fps) < 0.01;		
		bool half_speed		= std::abs(format_desc.fps / 2.0 - frame_factory_->get_video_format_desc().fps) < 0.01;

		if(half_speed && frame_number_ % 2 == 0) // Skip frame
			return receive(0);

		desc.pix_fmt = core::pixel_format::bgra;
		desc.planes.push_back(core::pixel_format_desc::plane(format_desc.width, format_desc.height, 4));
		auto frame = frame_factory_->create_frame(this, desc, read_frame->multichannel_view().channel_layout());

		bool copy_audio = !double_speed && !half_speed;

		if (copy_audio)
		{
			frame->audio_data().reserve(read_frame->audio_data().size());
			boost::copy(read_frame->audio_data(), std::back_inserter(frame->audio_data()));
		}

		fast_memcpy(frame->image_data().begin(), read_frame->image_data().begin(), read_frame->image_data().size());
		frame->commit();

		frame_buffer_.push(frame);	
		
		if(double_speed)	
			frame_buffer_.push(frame);

		return receive(0);
	}	

	virtual boost::unique_future<bool> send(const safe_ptr<core::read_frame>& frame) override
	{
		auto buffer = std::make_shared<audio_buffer_16>(
			core::audio_32_to_16(core::get_rearranged_and_mixed(frame->multichannel_view(), channel_layout_, channel_layout_.num_channels)));

		if (!input_.try_push(std::make_pair(frame, buffer)))
			graph_->set_tag("dropped-frame");

		if (Status() != Playing && !started_)
		{
			sf::SoundStream::Initialize(2, format_desc_.audio_sample_rate);
			Play();
			started_ = true;
		}

		return wrap_as_future(is_running_.load());
	}

	implementation(const safe_ptr<ogl_device>& ogl)
		: ogl_(ogl)
		, shader_(ogl_->invoke([&]{return get_image_shader(*ogl, blend_modes_, post_processing_);}))
		, supports_texture_barrier_(glTextureBarrierNV != 0)
	{
		if (!supports_texture_barrier_)
			CASPAR_LOG(warning) << L"[image_mixer] TextureBarrierNV not supported. Post processing will not be available";
	}

	void execute(const safe_ptr<read_frame>& frame)
	{			
		if(!has_synchronization_clock())
			timer_.tick(1.0/channel_.get_format_desc().fps);

		if(frame->image_size() != channel_.get_format_desc().size)
		{
			timer_.tick(1.0/channel_.get_format_desc().fps);
			return;
		}
		
		auto it = consumers_.begin();
		while(it != consumers_.end())
		{
			auto consumer = it->second;

			if(consumer->get_video_format_desc() != channel_.get_format_desc())
				consumer->initialize(channel_.get_format_desc());

			try
			{
				if(consumer->send(frame))
					++it;
				else
					consumers_.erase(it++);
			}
			catch(...)
			{
				CASPAR_LOG_CURRENT_EXCEPTION();
				CASPAR_LOG(warning) << "Trying to restart consumer: " << consumer->print() << L".";
				try
				{
					consumer->initialize(channel_.get_format_desc());
					consumer->send(frame);
				}
				catch(...)
				{	
					CASPAR_LOG_CURRENT_EXCEPTION();	
					CASPAR_LOG(warning) << "Consumer restart failed, trying to restart channel: " << consumer->print() << L".";	

					try
					{
						restart_channel_();
						consumer->initialize(channel_.get_format_desc());
						consumer->send(frame);
					}
					catch(...)
					{
						CASPAR_LOG_CURRENT_EXCEPTION();
						CASPAR_LOG(error) << "Failed to recover consumer: " << consumer->print() << L". Removing it.";
						consumers_.erase(it++);
					}
				}
			}
		}
	}

safe_ptr<basic_frame> receive_and_follow(safe_ptr<frame_producer>& producer, int hints)
{	
	auto frame = producer->receive(hints);
	if(frame == basic_frame::eof())
	{
		CASPAR_LOG(info) << producer->print() << " End Of File.";
		auto following = producer->get_following_producer();
		if(following != frame_producer::empty())
		{
			following->set_leading_producer(producer);
			producer = std::move(following);
		}
		else
			producer = make_safe<last_frame_producer>(producer);

		return receive_and_follow(producer, hints);
	}
	return frame;
}

	explicit transition_producer(const field_mode::type& mode, const safe_ptr<frame_producer>& dest, const transition_info& info) 
		: mode_(mode)
		, current_frame_(0)
		, info_(info)
		, dest_producer_(dest)
		, source_producer_(frame_producer::empty())
		, last_frame_(basic_frame::empty())
	{
		dest->monitor_output().link_target(&monitor_subject_);
	}

	virtual boost::unique_future<bool> send(const safe_ptr<core::read_frame>& frame) override
	{
		CASPAR_VERIFY(format_desc_.height * format_desc_.width * 4 == static_cast<unsigned>(frame->image_data().size()));

		return executor_.begin_invoke([=]() -> bool
		{			
			graph_->set_value("tick-time", tick_timer_.elapsed() * format_desc_.fps * 0.5);
			tick_timer_.restart();
			frame_timer_.restart();

			// AUDIO

			std::vector<int16_t, tbb::cache_aligned_allocator<int16_t>> audio_buffer;

			if (core::needs_rearranging(
					frame->multichannel_view(),
					channel_layout_,
					channel_layout_.num_channels))
			{
				core::audio_buffer downmixed;

				downmixed.resize(
						frame->multichannel_view().num_samples() 
								* channel_layout_.num_channels,
						0);

				auto dest_view = core::make_multichannel_view<int32_t>(
						downmixed.begin(), downmixed.end(), channel_layout_);

				core::rearrange_or_rearrange_and_mix(
						frame->multichannel_view(),
						dest_view,
						core::default_mix_config_repository());

				audio_buffer = core::audio_32_to_16(downmixed);
			}
			else
			{
				audio_buffer = core::audio_32_to_16(frame->audio_data());
			}

			airsend::add_audio(air_send_.get(), audio_buffer.data(), audio_buffer.size() / channel_layout_.num_channels);

			// VIDEO

			connected_ = airsend::add_frame_bgra(air_send_.get(), frame->image_data().begin());

			graph_->set_text(print());
			graph_->set_value("frame-time", frame_timer_.elapsed() * format_desc_.fps * 0.5);
			
			return true;
		});
	}

	implementation(video_channel& self, int index, const video_format_desc& format_desc, const safe_ptr<ogl_device>& ogl, const channel_layout& audio_channel_layout)  
		: self_(self)
		, index_(index)
		, format_desc_(format_desc)
		, ogl_(ogl)
		, output_(new caspar::core::output(graph_, format_desc, index))
		, mixer_(new caspar::core::mixer(graph_, output_, format_desc, ogl, audio_channel_layout))
		, stage_(new caspar::core::stage(graph_, mixer_, format_desc))	
		, monitor_subject_("/channel/" + boost::lexical_cast<std::string>(index))
	{
		graph_->set_text(print());
		diagnostics::register_graph(graph_);

		for(int n = 0; n < std::max(1, env::properties().get(L"configuration.pipeline-tokens", 2)); ++n)
			stage_->spawn_token();

		stage_->monitor_output().link_target(&monitor_subject_);

		CASPAR_LOG(info) << print() << " Successfully Initialized.";
	}

	virtual boost::unique_future<bool> send(const safe_ptr<read_frame>& frame) override
	{		
		if(audio_cadence_.size() == 1)
			return consumer_->send(frame);

		boost::unique_future<bool> result = caspar::wrap_as_future(true);
		
		if(boost::range::equal(sync_buffer_, audio_cadence_) && audio_cadence_.front() * frame->num_channels() == static_cast<size_t>(frame->audio_data().size()))
		{	
			// Audio sent so far is in sync, now we can send the next chunk.
			result = consumer_->send(frame);
			boost::range::rotate(audio_cadence_, std::begin(audio_cadence_)+1);
		}
		else
			CASPAR_LOG(trace) << print() << L" Syncing audio.";

		sync_buffer_.push_back(static_cast<size_t>(frame->audio_data().size() / frame->num_channels()));
		
		return std::move(result);
	}

	explicit layer_producer(const safe_ptr<frame_factory>& frame_factory, const safe_ptr<stage>& stage, int layer) 
		: frame_factory_(frame_factory)
		, stage_(stage)
		, consumer_(new layer_consumer())
		, last_frame_(basic_frame::empty())
		, frame_number_(0)
	{
		layer_ = layer;
		stage_->add_layer_consumer(this, layer_, consumer_);
		consumer_->block_until_first_frame_available();
		CASPAR_LOG(info) << print() << L" Initialized";
	}

clk_command_handler create_send_xml_handler(
    const std::wstring& command_name,
    bool expect_clock,
    bool expect_time,
    const safe_ptr<command_context>& context)
{
    return [=] (const std::vector<std::wstring>& params)
    {
        context->send_to_flash(get_xml(
                                   command_name, expect_clock, expect_time, params));
    };
}

	safe_ptr<basic_frame> receive(int hints)
	{		
		try
		{
			monitor_subject_ << monitor::message("/paused") % is_paused_;

			if(is_paused_)
			{
				if(foreground_->last_frame() == basic_frame::empty())
					foreground_->receive(frame_producer::NO_HINT);

				return disable_audio(foreground_->last_frame());
			}

			auto foreground = foreground_;

			auto frame = receive_and_follow(foreground, hints);

			if(foreground != foreground_)
				set_foreground(foreground);

			if(frame == core::basic_frame::late())
				return foreground_->last_frame();

			auto frames_left = static_cast<int64_t>(foreground_->nb_frames()) - static_cast<int64_t>(++frame_number_) - static_cast<int64_t>(auto_play_delta_);
			if(auto_play_delta_ > -1 && frames_left < 1)
			{
				play();
				return receive(hints);
			}
				
			return frame;
		}
		catch(...)
		{
			CASPAR_LOG_CURRENT_EXCEPTION();
			stop();
			return core::basic_frame::empty();
		}
	}