]> git.sesse.net Git - casparcg/blobdiff - modules/ffmpeg/consumer/streaming_consumer.cpp
[streaming_consumer] Implemented support for separating audio channels into separate...
[casparcg] / modules / ffmpeg / consumer / streaming_consumer.cpp
index 41cc3e79899fccc38679cffb8085f3caf384661a..f5d2997e8e010212f4921b5244c157d910e35c30 100644 (file)
@@ -4,15 +4,20 @@
 
 #include "../ffmpeg_error.h"
 #include "../producer/util/util.h"
+#include "../producer/filter/filter.h"
+#include "../producer/filter/audio_filter.h"
 
 #include <common/except.h>
 #include <common/executor.h>
 #include <common/assert.h>
 #include <common/utf.h>
 #include <common/future.h>
+#include <common/diagnostics/graph.h>
 #include <common/env.h>
 #include <common/scope_exit.h>
 #include <common/ptree.h>
+#include <common/param.h>
+#include <common/semaphore.h>
 
 #include <core/consumer/frame_consumer.h>
 #include <core/frame/frame.h>
@@ -62,29 +67,115 @@ extern "C"
 
 #pragma warning(pop)
 
-namespace caspar { namespace ffmpeg {
+namespace caspar { namespace ffmpeg { namespace {
 
-int crc16(const std::string& str)
+void set_pixel_format(AVFilterContext* sink, AVPixelFormat pix_fmt)
 {
-       boost::crc_16_type result;
+#pragma warning (push)
+#pragma warning (disable : 4245)
 
-       result.process_bytes(str.data(), str.length());
+       FF(av_opt_set_int_list(
+               sink,
+               "pix_fmts",
+               std::vector<AVPixelFormat>({ pix_fmt, AVPixelFormat::AV_PIX_FMT_NONE }).data(),
+               -1,
+               AV_OPT_SEARCH_CHILDREN));
 
-       return result.checksum();
+#pragma warning (pop)
 }
 
-class streaming_consumer final : public core::frame_consumer
+void adjust_video_filter(const AVCodec& codec, const core::video_format_desc& in_format, AVFilterContext* sink, std::string& filter)
 {
-public:
-       // Static Members
+       switch (codec.id)
+       {
+       case AV_CODEC_ID_DVVIDEO:
+               // Crop
+               if (in_format.format == core::video_format::ntsc)
+                       filter = u8(append_filter(u16(filter), L"crop=720:480:0:2"));
+
+               // Pixel format selection
+               if (in_format.format == core::video_format::ntsc)
+                       set_pixel_format(sink, AVPixelFormat::AV_PIX_FMT_YUV411P);
+               else if (in_format.format == core::video_format::pal)
+                       set_pixel_format(sink, AVPixelFormat::AV_PIX_FMT_YUV420P);
+               else
+                       set_pixel_format(sink, AVPixelFormat::AV_PIX_FMT_YUV422P);
+
+               // Scale
+               if (in_format.height == 1080)
+                       filter = u8(append_filter(u16(filter), in_format.duration == 1001
+                               ? L"scale=1280:1080"
+                               : L"scale=1440:1080"));
+               else if (in_format.height == 720)
+                       filter = u8(append_filter(u16(filter), L"scale=960:720"));
+
+               break;
+       }
+}
+
+void setup_codec_defaults(AVCodecContext& encoder)
+{
+       static const int MEGABIT = 1000000;
+
+       switch (encoder.codec_id)
+       {
+       case AV_CODEC_ID_DNXHD:
+               encoder.bit_rate = 220 * MEGABIT;
+
+               break;
+       case AV_CODEC_ID_PRORES:
+               encoder.bit_rate = encoder.width < 1280
+                               ?  63 * MEGABIT
+                               : 220 * MEGABIT;
+
+               break;
+       case AV_CODEC_ID_H264:
+               av_opt_set(encoder.priv_data,   "preset",       "ultrafast",    0);
+               av_opt_set(encoder.priv_data,   "tune",         "fastdecode",   0);
+               av_opt_set(encoder.priv_data,   "crf",          "5",                    0);
+
+               break;
+       }
+}
+
+bool is_pcm_s24le_not_supported(const AVFormatContext& container)
+{
+       auto name = std::string(container.oformat->name);
+
+       if (name == "mp4" || name == "dv")
+               return true;
+
+       return false;
+}
+
+template<typename Out, typename In>
+std::vector<Out> from_terminated_array(const In* array, In terminator)
+{
+       std::vector<Out> result;
+
+       while (array != nullptr && *array != terminator)
+       {
+               In val          = *array;
+               Out casted      = static_cast<Out>(val);
 
+               result.push_back(casted);
+
+               ++array;
+       }
+
+       return result;
+}
+
+class ffmpeg_consumer
+{
 private:
+       const spl::shared_ptr<diagnostics::graph>       graph_;
        core::monitor::subject                                          subject_;
-       boost::filesystem::path                                         path_;
-       int                                                                                     consumer_index_offset_;
+       std::string                                                                     path_;
+       boost::filesystem::path                                         full_path_;
 
        std::map<std::string, std::string>                      options_;
-       bool                                                                            compatibility_mode_;
+       bool                                                                            mono_streams_;
 
        core::video_format_desc                                         in_video_format_;
        core::audio_channel_layout                                      in_channel_layout_                      = core::audio_channel_layout::invalid();
@@ -93,45 +184,36 @@ private:
        tbb::atomic<bool>                                                       abort_request_;
 
        std::shared_ptr<AVStream>                                       video_st_;
-       std::shared_ptr<AVStream>                                       audio_st_;
+       std::vector<std::shared_ptr<AVStream>>          audio_sts_;
 
-       std::int64_t                                                            video_pts_;
-       std::int64_t                                                            audio_pts_;
+       std::int64_t                                                            video_pts_                                      = 0;
+       std::int64_t                                                            audio_pts_                                      = 0;
 
-    AVFilterContext*                                                   audio_graph_in_;
-    AVFilterContext*                                                   audio_graph_out_;
-    std::shared_ptr<AVFilterGraph>                             audio_graph_;
-       std::shared_ptr<AVBitStreamFilterContext>       audio_bitstream_filter_;
+       std::unique_ptr<audio_filter>                           audio_filter_;
 
+       // TODO: make use of already existent avfilter abstraction for video also
     AVFilterContext*                                                   video_graph_in_;
     AVFilterContext*                                                   video_graph_out_;
     std::shared_ptr<AVFilterGraph>                             video_graph_;
-       std::shared_ptr<AVBitStreamFilterContext>       video_bitstream_filter_;
-
-       executor                                                                        executor_;
 
        executor                                                                        video_encoder_executor_;
        executor                                                                        audio_encoder_executor_;
 
-       tbb::atomic<int>                                                        tokens_;
-       boost::mutex                                                            tokens_mutex_;
-       boost::condition_variable                                       tokens_cond_;
+       semaphore                                                                       tokens_                                         { 0 };
+
        tbb::atomic<int64_t>                                            current_encoding_delay_;
 
        executor                                                                        write_executor_;
 
 public:
 
-       streaming_consumer(
+       ffmpeg_consumer(
                        std::string path,
                        std::string options,
-                       bool compatibility_mode)
+                       bool mono_streams)
                : path_(path)
-               , consumer_index_offset_(crc16(path))
-               , compatibility_mode_(compatibility_mode)
-               , video_pts_(0)
-               , audio_pts_(0)
-               , executor_(print())
+               , full_path_(path)
+               , mono_streams_(mono_streams)
                , audio_encoder_executor_(print() + L" audio_encoder")
                , video_encoder_executor_(print() + L" video_encoder")
                , write_executor_(print() + L" io")
@@ -153,15 +235,15 @@ public:
         if (options_.find("threads") == options_.end())
             options_["threads"] = "auto";
 
-               tokens_ =
+               tokens_.release(
                        std::max(
                                1,
                                try_remove_arg<int>(
                                        options_,
-                                       boost::regex("tokens")).get_value_or(2));
+                                       boost::regex("tokens")).get_value_or(2)));
        }
 
-       ~streaming_consumer()
+       ~ffmpeg_consumer()
        {
                if(oc_)
                {
@@ -174,9 +256,9 @@ public:
                        audio_encoder_executor_.join();
 
                        video_graph_.reset();
-                       audio_graph_.reset();
+                       audio_filter_.reset();
                        video_st_.reset();
-                       audio_st_.reset();
+                       audio_sts_.clear();
 
                        write_packet(nullptr, nullptr);
 
@@ -194,39 +276,35 @@ public:
 
        void initialize(
                        const core::video_format_desc& format_desc,
-                       const core::audio_channel_layout& channel_layout,
-                       int channel_index) override
+                       const core::audio_channel_layout& channel_layout)
        {
                try
                {
                        static boost::regex prot_exp("^.+:.*" );
 
-                       const auto overwrite =
-                               try_remove_arg<std::string>(
-                                       options_,
-                                       boost::regex("y")) != boost::none;
-
                        if(!boost::regex_match(
-                                       path_.string(),
+                                       path_,
                                        prot_exp))
                        {
-                               if(!path_.is_complete())
+                               if(!full_path_.is_complete())
                                {
-                                       path_ =
+                                       full_path_ =
                                                u8(
                                                        env::media_folder()) +
-                                                       path_.string();
+                                                       path_;
                                }
 
-                               if(boost::filesystem::exists(path_))
-                               {
-                                       if(!overwrite && !compatibility_mode_)
-                                               BOOST_THROW_EXCEPTION(invalid_argument() << msg_info("File exists"));
+                               if(boost::filesystem::exists(full_path_))
+                                       boost::filesystem::remove(full_path_);
 
-                                       boost::filesystem::remove(path_);
-                               }
+                               boost::filesystem::create_directories(full_path_.parent_path());
                        }
 
+                       graph_->set_color("frame-time", diagnostics::color(0.1f, 1.0f, 0.1f));
+                       graph_->set_color("dropped-frame", diagnostics::color(0.3f, 0.6f, 0.3f));
+                       graph_->set_text(print());
+                       diagnostics::register_graph(graph_);
+
                        const auto oformat_name =
                                try_remove_arg<std::string>(
                                        options_,
@@ -238,7 +316,7 @@ public:
                                &oc,
                                nullptr,
                                oformat_name && !oformat_name->empty() ? oformat_name->c_str() : nullptr,
-                               path_.string().c_str()));
+                               full_path_.string().c_str()));
 
                        oc_.reset(
                                oc,
@@ -246,7 +324,7 @@ public:
 
                        CASPAR_VERIFY(oc_->oformat);
 
-                       oc_->interrupt_callback.callback = streaming_consumer::interrupt_cb;
+                       oc_->interrupt_callback.callback = ffmpeg_consumer::interrupt_cb;
                        oc_->interrupt_callback.opaque   = this;
 
                        CASPAR_VERIFY(format_desc.format != core::video_format::invalid);
@@ -274,7 +352,9 @@ public:
                        const auto audio_codec =
                                audio_codec_name
                                        ? avcodec_find_encoder_by_name(audio_codec_name->c_str())
-                                       : avcodec_find_encoder(oc_->oformat->audio_codec);
+                                       : (is_pcm_s24le_not_supported(*oc_)
+                                               ? avcodec_find_encoder(oc_->oformat->audio_codec)
+                                               : avcodec_find_encoder_by_name("pcm_s24le"));
 
                        if (!video_codec)
                                CASPAR_THROW_EXCEPTION(user_error() << msg_info(
@@ -303,12 +383,6 @@ public:
                                        boost::regex("af|f:a|filter:a")).get_value_or(""));
                        }
 
-                       // Bistream Filters
-                       {
-                               configue_audio_bistream_filters(options_);
-                               configue_video_bistream_filters(options_);
-                       }
-
                        // Encoders
 
                        {
@@ -317,11 +391,14 @@ public:
 
                                video_st_ = open_encoder(
                                        *video_codec,
-                                       video_options);
+                                       video_options,
+                                       0);
 
-                               audio_st_ = open_encoder(
-                                       *audio_codec,
-                                       audio_options);
+                               for (int i = 0; i < audio_filter_->get_num_output_pads(); ++i)
+                                       audio_sts_.push_back(open_encoder(
+                                                       *audio_codec,
+                                                       audio_options,
+                                                       i));
 
                                auto it = options_.begin();
                                while(it != options_.end())
@@ -350,7 +427,7 @@ public:
                                {
                                        FF(avio_open2(
                                                &oc_->pb,
-                                               path_.string().c_str(),
+                                               full_path_.string().c_str(),
                                                AVIO_FLAG_WRITE,
                                                &oc_->interrupt_callback,
                                                &av_opts));
@@ -383,90 +460,64 @@ public:
                catch(...)
                {
                        video_st_.reset();
-                       audio_st_.reset();
+                       audio_sts_.clear();
                        oc_.reset();
                        throw;
                }
        }
 
-       core::monitor::subject& monitor_output() override
+       core::monitor::subject& monitor_output()
        {
                return subject_;
        }
 
-       std::wstring name() const override
-       {
-               return L"streaming";
-       }
-
-       std::future<bool> send(core::const_frame frame) override
+       void send(core::const_frame frame)
        {
                CASPAR_VERIFY(in_video_format_.format != core::video_format::invalid);
 
-               --tokens_;
+               auto frame_timer = spl::make_shared<caspar::timer>();
+
                std::shared_ptr<void> token(
                        nullptr,
-                       [this, frame](void*)
+                       [this, frame, frame_timer](void*)
                        {
-                               ++tokens_;
-                               tokens_cond_.notify_one();
+                               tokens_.release();
                                current_encoding_delay_ = frame.get_age_millis();
+                               graph_->set_value("frame-time", frame_timer->elapsed() * in_video_format_.fps * 0.5);
                        });
+               tokens_.acquire();
 
-               return executor_.begin_invoke([=]() -> bool
+               video_encoder_executor_.begin_invoke([=]() mutable
                {
-                       boost::unique_lock<boost::mutex> tokens_lock(tokens_mutex_);
-
-                       while(tokens_ < 0)
-                               tokens_cond_.wait(tokens_lock);
-
-                       video_encoder_executor_.begin_invoke([=]() mutable
-                       {
-                               encode_video(
-                                       frame,
-                                       token);
-                       });
-
-                       audio_encoder_executor_.begin_invoke([=]() mutable
-                       {
-                               encode_audio(
-                                       frame,
-                                       token);
-                       });
-
-                       return true;
+                       encode_video(
+                               frame,
+                               token);
                });
-       }
-
-       std::wstring print() const override
-       {
-               return L"streaming_consumer[" + u16(path_.string()) + L"]";
-       }
 
-       virtual boost::property_tree::wptree info() const override
-       {
-               boost::property_tree::wptree info;
-               info.add(L"type", L"stream");
-               info.add(L"path", path_.wstring());
-               return info;
+               audio_encoder_executor_.begin_invoke([=]() mutable
+               {
+                       encode_audio(
+                               frame,
+                               token);
+               });
        }
 
-       bool has_synchronization_clock() const override
+       bool ready_for_frame() const
        {
-               return false;
+               return tokens_.permits() > 0;
        }
 
-       int buffer_depth() const override
+       void mark_dropped()
        {
-               return -1;
+               graph_->set_tag(diagnostics::tag_severity::WARNING, "dropped-frame");
        }
 
-       int index() const override
+       std::wstring print() const
        {
-               return compatibility_mode_ ? 200 : 100000 + consumer_index_offset_;
+               return L"ffmpeg_consumer[" + u16(path_) + L"]";
        }
 
-       int64_t presentation_frame_age_millis() const override
+       int64_t presentation_frame_age_millis() const
        {
                return current_encoding_delay_;
        }
@@ -476,13 +527,14 @@ private:
        static int interrupt_cb(void* ctx)
        {
                CASPAR_ASSERT(ctx);
-               return reinterpret_cast<streaming_consumer*>(ctx)->abort_request_;
+               return reinterpret_cast<ffmpeg_consumer*>(ctx)->abort_request_;
        }
 
        std::shared_ptr<AVStream> open_encoder(
                        const AVCodec& codec,
                        std::map<std::string,
-                       std::string>& options)
+                       std::string>& options,
+                       int stream_number_for_media_type)
        {
                auto st =
                        avformat_new_stream(
@@ -490,7 +542,7 @@ private:
                                &codec);
 
                if (!st)
-                       CASPAR_THROW_EXCEPTION(caspar_exception() << msg_info("Could not allocate video-stream.") << boost::errinfo_api_function("av_new_stream"));
+                       CASPAR_THROW_EXCEPTION(caspar_exception() << msg_info("Could not allocate video-stream.") << boost::errinfo_api_function("avformat_new_stream"));
 
                auto enc = st->codec;
 
@@ -501,8 +553,8 @@ private:
                        case AVMEDIA_TYPE_VIDEO:
                        {
                                enc->time_base                          = video_graph_out_->inputs[0]->time_base;
-                               enc->pix_fmt                            = static_cast<AVPixelFormat>(video_graph_out_->inputs[0]->format);
-                               enc->sample_aspect_ratio        = st->sample_aspect_ratio = video_graph_out_->inputs[0]->sample_aspect_ratio;
+                               enc->pix_fmt                                    = static_cast<AVPixelFormat>(video_graph_out_->inputs[0]->format);
+                               enc->sample_aspect_ratio                = st->sample_aspect_ratio = video_graph_out_->inputs[0]->sample_aspect_ratio;
                                enc->width                                      = video_graph_out_->inputs[0]->w;
                                enc->height                                     = video_graph_out_->inputs[0]->h;
                                enc->bit_rate_tolerance         = 400 * 1000000;
@@ -511,16 +563,18 @@ private:
                        }
                        case AVMEDIA_TYPE_AUDIO:
                        {
-                               enc->time_base                          = audio_graph_out_->inputs[0]->time_base;
-                               enc->sample_fmt                         = static_cast<AVSampleFormat>(audio_graph_out_->inputs[0]->format);
-                               enc->sample_rate                        = audio_graph_out_->inputs[0]->sample_rate;
-                               enc->channel_layout                     = audio_graph_out_->inputs[0]->channel_layout;
-                               enc->channels                           = audio_graph_out_->inputs[0]->channels;
+                               enc->time_base                          = audio_filter_->get_output_pad_info(stream_number_for_media_type).time_base;
+                               enc->sample_fmt                         = static_cast<AVSampleFormat>(audio_filter_->get_output_pad_info(stream_number_for_media_type).format);
+                               enc->sample_rate                                = audio_filter_->get_output_pad_info(stream_number_for_media_type).sample_rate;
+                               enc->channel_layout                     = audio_filter_->get_output_pad_info(stream_number_for_media_type).channel_layout;
+                               enc->channels                           = audio_filter_->get_output_pad_info(stream_number_for_media_type).channels;
 
                                break;
                        }
                }
 
+               setup_codec_defaults(*enc);
+
                if(oc_->oformat->flags & AVFMT_GLOBALHEADER)
                        enc->flags |= CODEC_FLAG_GLOBAL_HEADER;
 
@@ -576,8 +630,9 @@ private:
                if(enc->codec_type == AVMEDIA_TYPE_AUDIO && !(codec.capabilities & CODEC_CAP_VARIABLE_FRAME_SIZE))
                {
                        CASPAR_ASSERT(enc->frame_size > 0);
-                       av_buffersink_set_frame_size(audio_graph_out_,
-                                                                                enc->frame_size);
+                       audio_filter_->set_guaranteed_output_num_samples_per_frame(
+                                       stream_number_for_media_type,
+                                       enc->frame_size);
                }
 
                return std::shared_ptr<AVStream>(st, [this](AVStream* st)
@@ -586,61 +641,9 @@ private:
                });
        }
 
-       void configue_audio_bistream_filters(
-                       std::map<std::string, std::string>& options)
-       {
-               const auto audio_bitstream_filter_str =
-                       try_remove_arg<std::string>(
-                               options,
-                               boost::regex("^bsf:a|absf$"));
-
-               const auto audio_bitstream_filter =
-                       audio_bitstream_filter_str
-                               ? av_bitstream_filter_init(audio_bitstream_filter_str->c_str())
-                               : nullptr;
-
-               CASPAR_VERIFY(!audio_bitstream_filter_str || audio_bitstream_filter);
-
-               if(audio_bitstream_filter)
-               {
-                       audio_bitstream_filter_.reset(
-                               audio_bitstream_filter,
-                               av_bitstream_filter_close);
-               }
-
-               if(audio_bitstream_filter_str && !audio_bitstream_filter_)
-                       options["bsf:a"] = *audio_bitstream_filter_str;
-       }
-
-       void configue_video_bistream_filters(
-                       std::map<std::string, std::string>& options)
-       {
-               const auto video_bitstream_filter_str =
-                               try_remove_arg<std::string>(
-                                       options,
-                                       boost::regex("^bsf:v|vbsf$"));
-
-               const auto video_bitstream_filter =
-                       video_bitstream_filter_str
-                               ? av_bitstream_filter_init(video_bitstream_filter_str->c_str())
-                               : nullptr;
-
-               CASPAR_VERIFY(!video_bitstream_filter_str || video_bitstream_filter);
-
-               if(video_bitstream_filter)
-               {
-                       video_bitstream_filter_.reset(
-                               video_bitstream_filter,
-                               av_bitstream_filter_close);
-               }
-
-               if(video_bitstream_filter_str && !video_bitstream_filter_)
-                       options["bsf:v"] = *video_bitstream_filter_str;
-       }
-
        void configure_video_filters(
                        const AVCodec& codec,
-                       const std::string& filtergraph)
+                       std::string filtergraph)
        {
                video_graph_.reset(
                                avfilter_graph_alloc(),
@@ -662,7 +665,7 @@ private:
 
                const auto vsrc_options = (boost::format("video_size=%1%x%2%:pix_fmt=%3%:time_base=%4%/%5%:pixel_aspect=%6%/%7%:frame_rate=%8%/%9%")
                        % in_video_format_.width % in_video_format_.height
-                       % AV_PIX_FMT_BGRA
+                       % AVPixelFormat::AV_PIX_FMT_BGRA
                        % in_video_format_.duration     % in_video_format_.time_scale
                        % sample_aspect_ratio.numerator() % sample_aspect_ratio.denominator()
                        % in_video_format_.time_scale % in_video_format_.duration).str();
@@ -697,6 +700,8 @@ private:
 
 #pragma warning (pop)
 
+               adjust_video_filter(codec, in_video_format_, filt_vsink, filtergraph);
+
                configure_filtergraph(
                                *video_graph_,
                                filtergraph,
@@ -715,92 +720,42 @@ private:
 
        void configure_audio_filters(
                        const AVCodec& codec,
-                       const std::string& filtergraph)
+                       std::string filtergraph)
        {
-               audio_graph_.reset(
-                       avfilter_graph_alloc(),
-                       [](AVFilterGraph* p)
-                       {
-                               avfilter_graph_free(&p);
-                       });
-
-               audio_graph_->nb_threads  = boost::thread::hardware_concurrency()/2;
-               audio_graph_->thread_type = AVFILTER_THREAD_SLICE;
-
-               const auto asrc_options = (boost::format("sample_rate=%1%:sample_fmt=%2%:channels=%3%:time_base=%4%/%5%:channel_layout=%6%")
-                       % in_video_format_.audio_sample_rate
-                       % av_get_sample_fmt_name(AV_SAMPLE_FMT_S32)
-                       % in_channel_layout_.num_channels
-                       % 1     % in_video_format_.audio_sample_rate
-                       % boost::io::group(
-                               std::hex,
-                               std::showbase,
-                               av_get_default_channel_layout(in_channel_layout_.num_channels))).str();
+               int num_output_pads = 1;
 
-               AVFilterContext* filt_asrc = nullptr;
-               FF(avfilter_graph_create_filter(
-                       &filt_asrc,
-                       avfilter_get_by_name("abuffer"),
-                       "ffmpeg_consumer_abuffer",
-                       asrc_options.c_str(),
-                       nullptr,
-                       audio_graph_.get()));
-
-               AVFilterContext* filt_asink = nullptr;
-               FF(avfilter_graph_create_filter(
-                       &filt_asink,
-                       avfilter_get_by_name("abuffersink"),
-                       "ffmpeg_consumer_abuffersink",
-                       nullptr,
-                       nullptr,
-                       audio_graph_.get()));
-
-#pragma warning (push)
-#pragma warning (disable : 4245)
-
-               FF(av_opt_set_int(
-                       filt_asink,
-                       "all_channel_counts",
-                       1,
-                       AV_OPT_SEARCH_CHILDREN));
-
-               FF(av_opt_set_int_list(
-                       filt_asink,
-                       "sample_fmts",
-                       codec.sample_fmts,
-                       -1,
-                       AV_OPT_SEARCH_CHILDREN));
+               if (mono_streams_)
+               {
+                       num_output_pads = in_channel_layout_.num_channels;
+               }
 
-               FF(av_opt_set_int_list(
-                       filt_asink,
-                       "channel_layouts",
-                       codec.channel_layouts,
-                       -1,
-                       AV_OPT_SEARCH_CHILDREN));
+               if (num_output_pads > 1)
+               {
+                       std::string splitfilter = "[a:0]channelsplit=channel_layout=";
 
-               FF(av_opt_set_int_list(
-                       filt_asink,
-                       "sample_rates" ,
-                       codec.supported_samplerates,
-                       -1,
-                       AV_OPT_SEARCH_CHILDREN));
+                       splitfilter += (boost::format("0x%|1$x|") % create_channel_layout_bitmask(in_channel_layout_.num_channels)).str();
 
-#pragma warning (pop)
+                       for (int i = 0; i < num_output_pads; ++i)
+                               splitfilter += "[aout:" + boost::lexical_cast<std::string>(i) + "]";
 
-               configure_filtergraph(
-                       *audio_graph_,
-                       filtergraph,
-                       *filt_asrc,
-                       *filt_asink);
-
-               audio_graph_in_  = filt_asrc;
-               audio_graph_out_ = filt_asink;
+                       filtergraph = u8(append_filter(u16(filtergraph), u16(splitfilter)));
+               }
 
-               CASPAR_LOG(info)
-                       <<      u16(std::string("\n")
-                               + avfilter_graph_dump(
-                                       audio_graph_.get(),
-                                       nullptr));
+               std::vector<audio_output_pad> output_pads(
+                               num_output_pads,
+                               audio_output_pad(
+                                               from_terminated_array<int>(                             codec.supported_samplerates,    0),
+                                               from_terminated_array<AVSampleFormat>(  codec.sample_fmts,                              AVSampleFormat::AV_SAMPLE_FMT_NONE),
+                                               from_terminated_array<uint64_t>(                codec.channel_layouts,                  0ull)));
+
+               audio_filter_.reset(new audio_filter(
+                               { audio_input_pad(
+                                               boost::rational<int>(1, in_video_format_.audio_sample_rate),
+                                               in_video_format_.audio_sample_rate,
+                                               AVSampleFormat::AV_SAMPLE_FMT_S32,
+                                               create_channel_layout_bitmask(in_channel_layout_.num_channels)) },
+                                               output_pads,
+                                               filtergraph));
        }
 
        void configure_filtergraph(
@@ -812,51 +767,51 @@ private:
                AVFilterInOut* outputs = nullptr;
                AVFilterInOut* inputs = nullptr;
 
-               try
+               if(!filtergraph.empty())
                {
-                       if(!filtergraph.empty())
-                       {
-                               outputs = avfilter_inout_alloc();
-                               inputs  = avfilter_inout_alloc();
+                       outputs = avfilter_inout_alloc();
+                       inputs  = avfilter_inout_alloc();
 
+                       try
+                       {
                                CASPAR_VERIFY(outputs && inputs);
 
-                               outputs->name       = av_strdup("in");
-                               outputs->filter_ctx = &source_ctx;
-                               outputs->pad_idx    = 0;
-                               outputs->next       = nullptr;
+                               outputs->name           = av_strdup("in");
+                               outputs->filter_ctx     = &source_ctx;
+                               outputs->pad_idx                = 0;
+                               outputs->next           = nullptr;
 
-                               inputs->name        = av_strdup("out");
-                               inputs->filter_ctx  = &sink_ctx;
-                               inputs->pad_idx     = 0;
-                               inputs->next        = nullptr;
+                               inputs->name                    = av_strdup("out");
+                               inputs->filter_ctx      = &sink_ctx;
+                               inputs->pad_idx         = 0;
+                               inputs->next                    = nullptr;
+                       }
+                       catch (...)
+                       {
+                               avfilter_inout_free(&outputs);
+                               avfilter_inout_free(&inputs);
+                               throw;
+                       }
 
-                               FF(avfilter_graph_parse(
+                       FF(avfilter_graph_parse(
                                        &graph,
                                        filtergraph.c_str(),
                                        inputs,
                                        outputs,
                                        nullptr));
-                       }
-                       else
-                       {
-                               FF(avfilter_link(
+               }
+               else
+               {
+                       FF(avfilter_link(
                                        &source_ctx,
                                        0,
                                        &sink_ctx,
                                        0));
-                       }
+               }
 
-                       FF(avfilter_graph_config(
+               FF(avfilter_graph_config(
                                &graph,
                                nullptr));
-               }
-               catch(...)
-               {
-                       avfilter_inout_free(&outputs);
-                       avfilter_inout_free(&inputs);
-                       throw;
-               }
        }
 
        void encode_video(core::const_frame frame_ptr, std::shared_ptr<void> token)
@@ -878,15 +833,20 @@ private:
                                        in_video_format_.width,
                                        in_video_format_.height);
 
-                       src_av_frame->format                              AV_PIX_FMT_BGRA;
-                       src_av_frame->width                                       = in_video_format_.width;
-                       src_av_frame->height                              = in_video_format_.height;
-                       src_av_frame->sample_aspect_ratio.num = sample_aspect_ratio.numerator();
-                       src_av_frame->sample_aspect_ratio.den = sample_aspect_ratio.denominator();
-                       src_av_frame->pts                                         = video_pts_;
+                       src_av_frame->format                                            = AVPixelFormat::AV_PIX_FMT_BGRA;
+                       src_av_frame->width                                             = in_video_format_.width;
+                       src_av_frame->height                                            = in_video_format_.height;
+                       src_av_frame->sample_aspect_ratio.num   = sample_aspect_ratio.numerator();
+                       src_av_frame->sample_aspect_ratio.den   = sample_aspect_ratio.denominator();
+                       src_av_frame->pts                                               = video_pts_;
 
                        video_pts_ += 1;
 
+                       subject_
+                                       << core::monitor::message("/frame")     % video_pts_
+                                       << core::monitor::message("/path")      % path_
+                                       << core::monitor::message("/fps")       % in_video_format_.fps;
+
                        FF(av_image_fill_arrays(
                                src_av_frame->data,
                                src_av_frame->linesize,
@@ -919,7 +879,6 @@ private:
                                        {
                                                while(encode_av_frame(
                                                                *video_st_,
-                                                               video_bitstream_filter_.get(),
                                                                avcodec_encode_video2,
                                                                nullptr, token))
                                                {
@@ -948,7 +907,6 @@ private:
 
                                        encode_av_frame(
                                                *video_st_,
-                                               video_bitstream_filter_.get(),
                                                avcodec_encode_video2,
                                                filt_frame,
                                                token);
@@ -961,21 +919,19 @@ private:
 
        void encode_audio(core::const_frame frame_ptr, std::shared_ptr<void> token)
        {
-               if(!audio_st_)
+               if(audio_sts_.empty())
                        return;
 
-               auto enc = audio_st_->codec;
-
                if(frame_ptr != core::const_frame::empty())
                {
                        auto src_av_frame = create_frame();
 
-                       src_av_frame->channels           = in_channel_layout_.num_channels;
-                       src_av_frame->channel_layout = av_get_default_channel_layout(in_channel_layout_.num_channels);
-                       src_av_frame->sample_rate        = in_video_format_.audio_sample_rate;
-                       src_av_frame->nb_samples         = static_cast<int>(frame_ptr.audio_data().size()) / src_av_frame->channels;
-                       src_av_frame->format             = AV_SAMPLE_FMT_S32;
-                       src_av_frame->pts                        = audio_pts_;
+                       src_av_frame->channels                  = in_channel_layout_.num_channels;
+                       src_av_frame->channel_layout            = create_channel_layout_bitmask(in_channel_layout_.num_channels);
+                       src_av_frame->sample_rate               = in_video_format_.audio_sample_rate;
+                       src_av_frame->nb_samples                        = static_cast<int>(frame_ptr.audio_data().size()) / src_av_frame->channels;
+                       src_av_frame->format                            = AV_SAMPLE_FMT_S32;
+                       src_av_frame->pts                               = audio_pts_;
 
                        audio_pts_ += src_av_frame->nb_samples;
 
@@ -988,30 +944,40 @@ private:
                                        static_cast<AVSampleFormat>(src_av_frame->format),
                                        16));
 
-                       FF(av_buffersrc_add_frame(
-                                       audio_graph_in_,
-                                       src_av_frame.get()));
+                       audio_filter_->push(0, src_av_frame);
                }
 
-               int ret = 0;
-
-               while(ret >= 0)
+               for (int pad_id = 0; pad_id < audio_filter_->get_num_output_pads(); ++pad_id)
                {
-                       auto filt_frame = create_frame();
+                       for (auto filt_frame : audio_filter_->poll_all(pad_id))
+                       {
+                               audio_encoder_executor_.begin_invoke([=]
+                               {
+                                       encode_av_frame(
+                                                       *audio_sts_.at(pad_id),
+                                                       avcodec_encode_audio2,
+                                                       filt_frame,
+                                                       token);
 
-                       ret = av_buffersink_get_frame(
-                               audio_graph_out_,
-                               filt_frame.get());
+                                       boost::this_thread::yield(); // TODO:
+                               });
+                       }
+               }
+
+               bool eof = frame_ptr == core::const_frame::empty();
 
+               if (eof)
+               {
                        audio_encoder_executor_.begin_invoke([=]
                        {
-                               if(ret == AVERROR_EOF)
+                               for (int pad_id = 0; pad_id < audio_filter_->get_num_output_pads(); ++pad_id)
                                {
-                                       if(enc->codec->capabilities & CODEC_CAP_DELAY)
+                                       auto enc = audio_sts_.at(pad_id)->codec;
+
+                                       if (enc->codec->capabilities & CODEC_CAP_DELAY)
                                        {
-                                               while(encode_av_frame(
-                                                               *audio_st_,
-                                                               audio_bitstream_filter_.get(),
+                                               while (encode_av_frame(
+                                                               *audio_sts_.at(pad_id),
                                                                avcodec_encode_audio2,
                                                                nullptr,
                                                                token))
@@ -1020,21 +986,6 @@ private:
                                                }
                                        }
                                }
-                               else if(ret != AVERROR(EAGAIN))
-                               {
-                                       FF_RET(
-                                               ret,
-                                               "av_buffersink_get_frame");
-
-                                       encode_av_frame(
-                                               *audio_st_,
-                                               audio_bitstream_filter_.get(),
-                                               avcodec_encode_audio2,
-                                               filt_frame,
-                                               token);
-
-                                       boost::this_thread::yield(); // TODO:
-                               }
                        });
                }
        }
@@ -1042,7 +993,6 @@ private:
        template<typename F>
        bool encode_av_frame(
                        AVStream& st,
-                       AVBitStreamFilterContext* bsfc,
                        const F& func,
                        const std::shared_ptr<AVFrame>& src_av_frame,
                        std::shared_ptr<void> token)
@@ -1063,62 +1013,6 @@ private:
 
                pkt.stream_index = st.index;
 
-               if(bsfc)
-               {
-                       auto new_pkt = pkt;
-
-                       auto a = av_bitstream_filter_filter(
-                                       bsfc,
-                                       st.codec,
-                                       nullptr,
-                                       &new_pkt.data,
-                                       &new_pkt.size,
-                                       pkt.data,
-                                       pkt.size,
-                                       pkt.flags & AV_PKT_FLAG_KEY);
-
-                       if(a == 0 && new_pkt.data != pkt.data && new_pkt.destruct)
-                       {
-                               auto t = reinterpret_cast<std::uint8_t*>(av_malloc(new_pkt.size + FF_INPUT_BUFFER_PADDING_SIZE));
-
-                               if(t)
-                               {
-                                       memcpy(
-                                               t,
-                                               new_pkt.data,
-                                               new_pkt.size);
-
-                                       memset(
-                                               t + new_pkt.size,
-                                               0,
-                                               FF_INPUT_BUFFER_PADDING_SIZE);
-
-                                       new_pkt.data = t;
-                                       new_pkt.buf  = nullptr;
-                               }
-                               else
-                                       a = AVERROR(ENOMEM);
-                       }
-
-                       av_free_packet(&pkt);
-
-                       FF_RET(
-                               a,
-                               "av_bitstream_filter_filter");
-
-                       new_pkt.buf =
-                               av_buffer_create(
-                                       new_pkt.data,
-                                       new_pkt.size,
-                                       av_buffer_default_free,
-                                       nullptr,
-                                       0);
-
-                       CASPAR_VERIFY(new_pkt.buf);
-
-                       pkt = new_pkt;
-               }
-
                if (pkt.pts != AV_NOPTS_VALUE)
                {
                        pkt.pts =
@@ -1246,37 +1140,179 @@ private:
        }
 };
 
+int crc16(const std::string& str)
+{
+       boost::crc_16_type result;
+
+       result.process_bytes(str.data(), str.length());
+
+       return result.checksum();
+}
+
+struct ffmpeg_consumer_proxy : public core::frame_consumer
+{
+       const std::string                                       path_;
+       const std::string                                       options_;
+       const bool                                                      separate_key_;
+       const bool                                                      mono_streams_;
+       const bool                                                      compatibility_mode_;
+       int                                                                     consumer_index_offset_;
+
+       std::unique_ptr<ffmpeg_consumer>        consumer_;
+       std::unique_ptr<ffmpeg_consumer>        key_only_consumer_;
+
+public:
+
+       ffmpeg_consumer_proxy(const std::string& path, const std::string& options, bool separate_key, bool mono_streams, bool compatibility_mode)
+               : path_(path)
+               , options_(options)
+               , separate_key_(separate_key)
+               , mono_streams_(mono_streams)
+               , compatibility_mode_(compatibility_mode)
+               , consumer_index_offset_(crc16(path))
+       {
+       }
+
+       void initialize(const core::video_format_desc& format_desc, const core::audio_channel_layout& channel_layout, int) override
+       {
+               if (consumer_)
+                       CASPAR_THROW_EXCEPTION(invalid_operation() << msg_info("Cannot reinitialize ffmpeg-consumer."));
+
+               consumer_.reset(new ffmpeg_consumer(path_, options_, mono_streams_));
+               consumer_->initialize(format_desc, channel_layout);
+
+               if (separate_key_)
+               {
+                       boost::filesystem::path fill_file(path_);
+                       auto without_extension = u16(fill_file.parent_path().string() + "/" + fill_file.stem().string());
+                       auto key_file = without_extension + L"_A" + u16(fill_file.extension().string());
+
+                       key_only_consumer_.reset(new ffmpeg_consumer(u8(key_file), options_, mono_streams_));
+                       key_only_consumer_->initialize(format_desc, channel_layout);
+               }
+       }
+
+       int64_t presentation_frame_age_millis() const override
+       {
+               return consumer_ ? static_cast<int64_t>(consumer_->presentation_frame_age_millis()) : 0;
+       }
+
+       std::future<bool> send(core::const_frame frame) override
+       {
+               bool ready_for_frame = consumer_->ready_for_frame();
+
+               if (ready_for_frame && separate_key_)
+                       ready_for_frame = ready_for_frame && key_only_consumer_->ready_for_frame();
+
+               if (ready_for_frame)
+               {
+                       consumer_->send(frame);
+
+                       if (separate_key_)
+                               key_only_consumer_->send(frame.key_only());
+               }
+               else
+               {
+                       consumer_->mark_dropped();
+
+                       if (separate_key_)
+                               key_only_consumer_->mark_dropped();
+               }
+
+               return make_ready_future(true);
+       }
+
+       std::wstring print() const override
+       {
+               return consumer_ ? consumer_->print() : L"[ffmpeg_consumer]";
+       }
+
+       std::wstring name() const override
+       {
+               return L"ffmpeg";
+       }
+
+       boost::property_tree::wptree info() const override
+       {
+               boost::property_tree::wptree info;
+
+               info.add(L"type",                       L"ffmpeg");
+               info.add(L"path",                       u16(path_));
+               info.add(L"separate_key",       separate_key_);
+               info.add(L"mono_streams",       mono_streams_);
+
+               return info;
+       }
+
+       bool has_synchronization_clock() const override
+       {
+               return false;
+       }
+
+       int buffer_depth() const override
+       {
+               return -1;
+       }
+
+       int index() const override
+       {
+               return compatibility_mode_ ? 200 : 100000 + consumer_index_offset_;
+       }
+
+       core::monitor::subject& monitor_output() override
+       {
+               return consumer_->monitor_output();
+       }
+};
+
+}
+
 void describe_streaming_consumer(core::help_sink& sink, const core::help_repository& repo)
 {
-       sink.short_description(L"For streaming the contents of a channel using FFmpeg.");
-       sink.syntax(L"STREAM [url:string] {-[ffmpeg_param1:string] [value1:string] {-[ffmpeg_param2:string] [value2:string] {...}}}");
-       sink.para()->text(L"For streaming the contents of a channel using FFmpeg");
+       sink.short_description(L"For streaming/recording the contents of a channel using FFmpeg.");
+       sink.syntax(L"FILE,STREAM [filename:string],[url:string] {-[ffmpeg_param1:string] [value1:string] {-[ffmpeg_param2:string] [value2:string] {...}}} {[separate_key:SEPARATE_KEY]} {[mono_streams:MONO_STREAMS]}");
+       sink.para()->text(L"For recording or streaming the contents of a channel using FFmpeg");
        sink.definitions()
-               ->item(L"url", L"The stream URL to create/stream to.")
-               ->item(L"ffmpeg_paramX", L"A parameter supported by FFmpeg. For example vcodec or acodec etc.");
+               ->item(L"filename",                     L"The filename under the media folder including the extension (decides which kind of container format that will be used).")
+               ->item(L"url",                          L"If the filename is given in the form of an URL a network stream will be created instead of a file on disk.")
+               ->item(L"ffmpeg_paramX",                L"A parameter supported by FFmpeg. For example vcodec or acodec etc.")
+               ->item(L"separate_key",         L"If defined will create two files simultaneously -- One for fill and one for key (_A will be appended).")
+               ->item(L"mono_streams",         L"If defined every audio channel will be written to its own audio stream.");
        sink.para()->text(L"Examples:");
-       sink.example(L">> ADD 1 STREAM udp://<client_ip_address>:9250 -format mpegts -vcodec libx264 -crf 25 -tune zerolatency -preset ultrafast");
+       sink.example(L">> ADD 1 FILE output.mov -vcodec dnxhd");
+       sink.example(L">> ADD 1 FILE output.mov -vcodec prores");
+       sink.example(L">> ADD 1 FILE output.mov -vcodec dvvideo");
+       sink.example(L">> ADD 1 FILE output.mov -vcodec libx264 -preset ultrafast -tune fastdecode -crf 25");
+       sink.example(L">> ADD 1 FILE output.mov -vcodec dnxhd SEPARATE_KEY", L"for creating output.mov with fill and output_A.mov with key/alpha");
+       sink.example(L">> ADD 1 FILE output.mxf -vcodec dnxhd MONO_STREAMS", L"for creating output.mxf with every audio channel encoded in its own mono stream.");
+       sink.example(L">> ADD 1 STREAM udp://<client_ip_address>:9250 -format mpegts -vcodec libx264 -crf 25 -tune zerolatency -preset ultrafast",
+               L"for streaming over UDP instead of creating a local file.");
 }
 
 spl::shared_ptr<core::frame_consumer> create_streaming_consumer(
-               const std::vector<std::wstring>& params, core::interaction_sink*)
+               const std::vector<std::wstring>& params, core::interaction_sink*, std::vector<spl::shared_ptr<core::video_channel>> channels)
 {
        if (params.size() < 1 || (!boost::iequals(params.at(0), L"STREAM") && !boost::iequals(params.at(0), L"FILE")))
                return core::frame_consumer::empty();
 
-       auto compatibility_mode = boost::iequals(params.at(0), L"FILE");
-       auto path = u8(params.size() > 1 ? params.at(1) : L"");
-       auto args = u8(boost::join(params, L" "));
+       auto params2                    = params;
+       bool separate_key               = get_and_consume_flag(L"SEPARATE_KEY", params2);
+       bool mono_streams               = get_and_consume_flag(L"MONO_STREAMS", params2);
+       auto compatibility_mode = boost::iequals(params.at(0), L"FILE");
+       auto path                               = u8(params2.size() > 1 ? params2.at(1) : L"");
+       auto args                               = u8(boost::join(params2, L" "));
 
-       return spl::make_shared<streaming_consumer>(path, args, compatibility_mode);
+       return spl::make_shared<ffmpeg_consumer_proxy>(path, args, separate_key, mono_streams, compatibility_mode);
 }
 
 spl::shared_ptr<core::frame_consumer> create_preconfigured_streaming_consumer(
-               const boost::property_tree::wptree& ptree, core::interaction_sink*)
+               const boost::property_tree::wptree& ptree, core::interaction_sink*, std::vector<spl::shared_ptr<core::video_channel>> channels)
 {
-       return spl::make_shared<streaming_consumer>(
+       return spl::make_shared<ffmpeg_consumer_proxy>(
                        u8(ptree_get<std::wstring>(ptree, L"path")),
                        u8(ptree.get<std::wstring>(L"args", L"")),
+                       ptree.get<bool>(L"separate-key", false),
+                       ptree.get<bool>(L"mono-streams", false),
                        false);
 }