// Copyright 2019 Fuzhou Rockchip Electronics Co., Ltd. All rights reserved.
// Use of this source code is governed by a BSD-style license that can be
// found in the LICENSE file.

#include "flow.h"

#include <algorithm>
#include <assert.h>
#include <sys/prctl.h>

#include "buffer.h"
#include "key_string.h"
#include "utils.h"

namespace easymedia
{

    class FlowCoroutine
    {
      public:
        FlowCoroutine(Flow* f, Model sync_model, FunctionProcess func, float inter);
        ~FlowCoroutine();

        void Bind(std::vector<int>& in, std::vector<int>& out);
        bool Start();
        void RunOnce();
        int GetCachedBufferCnt();

      private:
        void WhileRun();
        void WhileRunSleep();
        void SyncFetchInput(MediaBufferVector& in);
        void ASyncFetchInputCommon(MediaBufferVector& in);
        void ASyncFetchInputAtomic(MediaBufferVector& in);

        void SendNullBufferDown(Flow::FlowMap& fm, const MediaBufferVector& in, std::list<Flow::FlowInputMap>& flows);
        void SendBufferDown(Flow::FlowMap& fm, const MediaBufferVector& in, std::list<Flow::FlowInputMap>& flows,
                            bool process_ret);
        void SendBufferDownFromDeque(Flow::FlowMap& fm, const MediaBufferVector& in,
                                     std::list<Flow::FlowInputMap>& flows, bool process_ret);
        size_t OutputHoldRelated(Flow::FlowMap& fm, std::shared_ptr<MediaBuffer>& out_buffer,
                                 const MediaBufferVector& input_vector);

        Flow* flow;
        Model model;
        float interval;
        std::vector<int> in_slots;
        std::vector<int> out_slots;
        std::thread* th;
        FunctionProcess th_run;

        MediaBufferVector in_vector;
        decltype(&FlowCoroutine::SyncFetchInput) fetch_input_func;
        decltype(&FlowCoroutine::SendBufferDown) send_down_func;

      public:
        void SetMarkName(std::string s)
        {
            name = s;
        }
        void SetExpectProcessTime(int time)
        {
            expect_process_time = time;
        }

        std::string name;
        int expect_process_time; // ms
    };

    FlowCoroutine::FlowCoroutine(Flow* f, Model sync_model, FunctionProcess func, float inter)
        : flow(f), model(sync_model), interval(inter), th(nullptr), th_run(func), expect_process_time(0)

    {
    }

    FlowCoroutine::~FlowCoroutine()
    {
        if (th) {
            th->join();
            delete th;
        }
        LOG("%s quit\n", name.c_str());
    }

    void FlowCoroutine::Bind(std::vector<int>& in, std::vector<int>& out)
    {
        assert(in_slots.size() == 0 && out_slots.size() == 0 && "flow coroutine binded");
        in_slots = in;
        out_slots = out;
    }

    bool FlowCoroutine::Start()
    {
        bool need_thread = false;
        auto func = &FlowCoroutine::WhileRun;
        switch (model) {
            case Model::ASYNCCOMMON:
                need_thread = true;
                fetch_input_func = &FlowCoroutine::ASyncFetchInputCommon;
                send_down_func = &FlowCoroutine::SendBufferDownFromDeque;
                break;
            case Model::ASYNCATOMIC:
                need_thread = true;
                fetch_input_func = &FlowCoroutine::ASyncFetchInputAtomic;
                send_down_func = &FlowCoroutine::SendBufferDown;
                func = &FlowCoroutine::WhileRunSleep;
                break;
            case Model::SYNC:
                fetch_input_func = &FlowCoroutine::SyncFetchInput;
                send_down_func = &FlowCoroutine::SendBufferDown;
                break;
            default:
                LOG("invalid model %d\n", (int)model);
                return false;
        }
        in_vector.resize(in_slots.size());
        if (need_thread) {
            th = new std::thread(func, this);
            if (!th) {
                errno = ENOMEM;
                return false;
            }
        }
        return true;
    }

#ifndef NDEBUG
    static void check_consume_time(const char* name, int expect, int exactly)
    {
        if (exactly > expect) {
            LOG("%s, expect consume %d ms, however %d ms\n", name, expect, exactly);
        }
    }
#endif

    void FlowCoroutine::RunOnce()
    {
        bool ret = true;
        (this->*fetch_input_func)(in_vector);

        if (flow->GetRunTimesRemaining()) {
#ifndef NDEBUG
            {
                AutoDuration ad;
#endif
                ret = (*th_run)(flow, in_vector);
#ifndef NDEBUG
                if (expect_process_time > 0)
                    check_consume_time(name.c_str(), expect_process_time, (int)(ad.Get() / 1000));
            }
#endif // DEBUG
        }

        for (int idx : out_slots) {
            auto& fm = flow->downflowmap[idx];
            std::list<Flow::FlowInputMap> flows;
            fm.list_mtx.read_lock();
            flows = fm.flows;
            fm.list_mtx.unlock();
            (this->*send_down_func)(fm, in_vector, flows, ret);
        }
        for (auto& buffer : in_vector) {
            buffer.reset();
        }
    }

    void FlowCoroutine::WhileRun()
    {
        prctl(PR_SET_NAME, this->name.c_str());
        LOGD("flow-name %s\n", this->name.c_str());
        while (!flow->quit) {
            RunOnce();
        }
    }

    void FlowCoroutine::WhileRunSleep()
    {
        int64_t times = 0;
        AutoDuration ad;
        assert(interval > 0);
        prctl(PR_SET_NAME, this->name.c_str());
        LOGD("flow-name %s\n", this->name.c_str());

        while (!flow->quit) {
            if (times == 0) {
                ad.Reset();
            }
            RunOnce();
            ++times;
            int64_t remain = (interval * times - ad.Get()) / 1000;
            if (remain > 0) {
                msleep((int)remain);
            }
            if (times >= 10000) {
                times = 0;
            }
        }
    }

    void FlowCoroutine::SyncFetchInput(MediaBufferVector& in)
    {
        int i = 0;
        for (int idx : in_slots) {
            auto& buffer = flow->v_input[idx].cached_buffer;
            in[i++] = buffer;
            buffer.reset();
        }
    }

    void FlowCoroutine::ASyncFetchInputCommon(MediaBufferVector& in)
    {
        flow->cond_mtx.lock();
        bool empty = true;
        for (size_t i = 0; i < in_slots.size(); i++) {
            int idx = in_slots[i];
            auto& input = flow->v_input[idx];
            auto& v = input.cached_buffers;
            if (!v.empty()) {
                empty = false;
            }
        }
        if (empty) {
            flow->cond_mtx.wait();
        }
        flow->cond_mtx.unlock();

        for (size_t i = 0; i < in_slots.size(); i++) {
            int idx = in_slots[i];
            auto& input = flow->v_input[idx];
            auto& v = input.cached_buffers;
            if (v.empty()) {
                continue;
            }
            if (!flow->enable) {
                in.assign(in_slots.size(), nullptr);
                break;
            }
            AutoLockMutex _alm(input.mtx);
            assert(!v.empty());
            in[i] = v.front();
            v.pop_front();
        }
    }

    void FlowCoroutine::ASyncFetchInputAtomic(MediaBufferVector& in)
    {
        int i = 0;
        for (int idx : in_slots) {
            std::shared_ptr<MediaBuffer> buffer;
            auto& input = flow->v_input[idx];
            input.spin_mtx.lock();
            buffer = input.cached_buffer;
            input.spin_mtx.unlock();
            in[i++] = buffer;
        }
    }

    void FlowCoroutine::SendNullBufferDown(Flow::FlowMap& fm, const MediaBufferVector& in,
                                           std::list<Flow::FlowInputMap>& flows)
    {
        std::shared_ptr<MediaBuffer> nullbuffer;
        if (fm.hold_input != HoldInputMode::NONE) {
            auto empty_result = std::make_shared<easymedia::MediaBuffer>();
            if (empty_result && OutputHoldRelated(fm, empty_result, in) > 0) {
                nullbuffer = empty_result;
            }
        }
        for (auto& f : flows) {
            f.flow->SendInput(nullbuffer, f.index_of_in);
        }
    }

    void FlowCoroutine::SendBufferDown(Flow::FlowMap& fm, const MediaBufferVector& in,
                                       std::list<Flow::FlowInputMap>& flows, bool process_ret)
    {
        if (!process_ret) {
            SendNullBufferDown(fm, in, flows);
            return;
        }
        for (auto& f : flows) {
            OutputHoldRelated(fm, fm.cached_buffer, in);
            f.flow->SendInput(fm.cached_buffer, f.index_of_in);
        }
        fm.cached_buffer.reset();
    }

    void FlowCoroutine::SendBufferDownFromDeque(Flow::FlowMap& fm, const MediaBufferVector& in,
                                                std::list<Flow::FlowInputMap>& flows, bool process_ret)
    {
        if (!process_ret) {
            SendNullBufferDown(fm, in, flows);
            return;
        }
        if (fm.cached_buffers.empty()) {
            return;
        }
        for (auto& buffer : fm.cached_buffers) {
            OutputHoldRelated(fm, buffer, in);
            for (auto& f : flows) {
                f.flow->SendInput(buffer, f.index_of_in);
            }
        }
        fm.cached_buffers.clear();
    }

    size_t FlowCoroutine::OutputHoldRelated(Flow::FlowMap& fm, std::shared_ptr<MediaBuffer>& out_buffer,
                                            const MediaBufferVector& input_vector)
    {
        if (fm.hold_input == HoldInputMode::HOLD_INPUT) {
            return FlowOutputHoldInput(out_buffer, input_vector);
        } else if (fm.hold_input == HoldInputMode::INHERIT_FORM_INPUT) {
            return FlowOutputInheritFromInput(out_buffer, input_vector);
        }
        return 0;
    }

    int FlowCoroutine::GetCachedBufferCnt()
    {
        int cnt = 0;
        for (auto inv : in_vector) {
            if (inv) {
                cnt++;
            }
        }

        return cnt;
    }

    DEFINE_REFLECTOR(Flow)
    DEFINE_FACTORY_COMMON_PARSE(Flow)
    DEFINE_PART_FINAL_EXPOSE_PRODUCT(Flow, Flow)

    const FunctionProcess Flow::void_transaction00 = void_transaction<0, 0>;

    Flow::Flow()
        : out_slot_num(0), input_slot_num(0), down_flow_num(0), waite_down_flow(true), event_handler2_(nullptr),
          event_callback_(nullptr), enable(true), quit(false), event_handler_(nullptr), play_video_handler_(nullptr),
          play_audio_handler_(nullptr), user_handler_(nullptr), user_callback_(nullptr), out_handler_(nullptr),
          out_callback_(nullptr), run_times(-1)
    {
    }

    Flow::~Flow()
    {
        StopAllThread();
    }

    void Flow::StopAllThread()
    {
        enable = false;
        quit = true;
        cond_mtx.lock();
        cond_mtx.notify();
        cond_mtx.unlock();
        for (auto& coroutine : coroutines) {
            coroutine.reset();
        }
    }

    bool Flow::IsAllBuffEmpty()
    {
#ifndef NDEBUG
        int i = 0;

        for (auto& input : v_input) {
            LOG("#FLOW v_input-%d cached_buffers size:%zu\n", i, input.cached_buffers.size());
            LOG("#FLOW v_input-%d cached_buffer :%s\n", i++, input.cached_buffer ? "NotNull" : "Null");
        }

        i = 0;
        for (auto& coroutin : coroutines) {
            LOG("#FLOW coroutin-%d in_vector size:%d\n", i++, coroutin->GetCachedBufferCnt());
        }

        i = 0;
        for (auto& fm : downflowmap) {
            LOG("#FLOW downflowmap-%d cached_buffers size:%zu\n", i, fm.cached_buffers.size());
            LOG("#FLOW downflowmap-%d cached_buffer : %s\n", i++, fm.cached_buffer ? "NotNull" : "Null");
        }
#endif

        for (auto& input : v_input) {
            if (input.cached_buffers.size() || input.cached_buffer) {
                return false;
            }
        }

        for (auto& coroutin : coroutines) {
            if (coroutin->GetCachedBufferCnt()) {
                return false;
            }
        }

        for (auto& fm : downflowmap) {
            if (fm.cached_buffers.size() || fm.cached_buffer) {
                return false;
            }
        }

        return true;
    }

    void Flow::StartStream()
    {
        waite_down_flow = false;
        if (source_start_cond_mtx) {
            source_start_cond_mtx->lock();
            source_start_cond_mtx->notify();
            source_start_cond_mtx->unlock();
        }
    }

    int Flow::SetRunTimes(int _run_times)
    {
        run_times = _run_times;
        LOG("Flow:%s set run_times to %d\n", flow_tag.c_str(), run_times);
        return run_times;
    }

    int Flow::GetRunTimesRemaining()
    {
        int remaining_value = run_times;

        if (run_times > 0) {
            run_times--;
        }

        return remaining_value;
    }

    void Flow::DumpBase(std::string& dump_info)
    {
        int idx = 0;
        char str_line[1024] = {0};

        dump_info = "";
        sprintf(str_line, "#Dump Flow(%s) base info:\r\n", GetFlowTag());
        dump_info.append(str_line);
        memset(str_line, 0, sizeof(str_line));
        sprintf(str_line, "  InSlotNum: %d\r\n", input_slot_num);
        dump_info.append(str_line);
        memset(str_line, 0, sizeof(str_line));
        sprintf(str_line, "  OutSlotNum: %d\r\n", out_slot_num);
        dump_info.append(str_line);

        for (auto& input : v_input) {
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "  ->Input[%d] info:\r\n", idx);
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "    Valid: %s\r\n", input.valid ? "True" : "False");
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            if (input.thread_model == Model::ASYNCCOMMON) {
                sprintf(str_line, "    ThreadMode: ASYNCCOMMON\r\n");
            } else if (input.thread_model == Model::ASYNCATOMIC) {
                sprintf(str_line, "    ThreadMode: ASYNCATOMIC\r\n");
            } else if (input.thread_model == Model::SYNC) {
                sprintf(str_line, "    ThreadMode: SYNC\r\n");
            } else {
                sprintf(str_line, "    ThreadMode: NONE\r\n");
            }
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            if (input.mode_when_full == InputMode::BLOCKING) {
                sprintf(str_line, "    InputMode: BLOCKING\r\n");
            } else if (input.mode_when_full == InputMode::DROPCURRENT) {
                sprintf(str_line, "    InputMode: DROPCURRENT\r\n");
            } else if (input.mode_when_full == InputMode::DROPFRONT) {
                sprintf(str_line, "    InputMode: DROPFRONT\r\n");
            } else {
                sprintf(str_line, "    InputMode: NONE\r\n");
            }
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "    BufferCnt: current:%d, max:%d\r\n", input.cached_buffers.size(),
                    input.max_cache_num);
            dump_info.append(str_line);
        }

        idx = 0;
        for (auto& fm : downflowmap) {
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "  ->FlowMap[%d] info:\r\n", idx);
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "    Valid: %s\r\n", fm.valid ? "True" : "False");
            dump_info.append(str_line);
            memset(str_line, 0, sizeof(str_line));
            sprintf(str_line, "    BufferCnt: %d\r\n", fm.cached_buffers.size());
            dump_info.append(str_line);

            dump_info.append("    NextFlow: ");
            for (auto& nflow : fm.flows) {
                dump_info.append(nflow.flow->GetFlowTag());
                dump_info.append(" ");
            }
            dump_info.append("\r\n");
        }
    }

    static bool check_slots(std::vector<int>& slots, const char* debugstr)
    {
        if (slots.empty()) {
            return true;
        }
        std::sort(slots.begin(), slots.end());
        auto iend = std::unique(slots.begin(), slots.end());
        if (iend != slots.end()) {
            LOG("%s slot duplicate :", debugstr);
            while (iend != slots.end()) {
                LOG(" %d", *iend);
                iend++;
            }
            LOG("\n");
            return false;
        }
        return true;
    }

    Flow::FlowMap::FlowMap(FlowMap&& fm)
    {
        if (fm.valid) {
            LOG("Flow::FlowMap is not copyable and moveable after inited\n");
            assert(0);
        }
    }

    void Flow::FlowMap::Init(Model m, HoldInputMode hold_in)
    {
        assert(!valid);
        valid = true;
        if (m == Model::ASYNCCOMMON) {
            set_output_behavior = &FlowMap::SetOutputToQueueBehavior;
        } else {
            set_output_behavior = &FlowMap::SetOutputBehavior;
        }
        hold_input = hold_in;
    }

    void Flow::FlowMap::SetOutputBehavior(const std::shared_ptr<MediaBuffer>& output)
    {
        cached_buffer = output;
    }
    void Flow::FlowMap::SetOutputToQueueBehavior(const std::shared_ptr<MediaBuffer>& output)
    {
        cached_buffers.push_back(output);
    }

    Flow::Input::Input(Input&& in)
    {
        if (in.valid) {
            LOG("Flow::Input is not copyable and moveable after inited\n");
            assert(0);
        }
    }

    void Flow::Input::Init(Flow* f, Model m, int mcn, InputMode im, bool f_block, std::shared_ptr<FlowCoroutine> fc)
    {
        assert(!valid);
        valid = true;
        flow = f;
        thread_model = m;
        fetch_block = f_block;
        max_cache_num = mcn;
        mode_when_full = im;
        switch (m) {
            case Model::ASYNCCOMMON:
                send_input_behavior = &Input::ASyncSendInputCommonBehavior;
                break;
            case Model::ASYNCATOMIC:
                send_input_behavior = &Input::ASyncSendInputAtomicBehavior;
                break;
            case Model::SYNC:
                send_input_behavior = &Input::SyncSendInputBehavior;
                coroutine = fc;
                break;
            default:
                break;
        }
        switch (im) {
            case InputMode::BLOCKING:
                async_full_behavior = &Input::ASyncFullBlockingBehavior;
                break;
            case InputMode::DROPFRONT:
                async_full_behavior = &Input::ASyncFullDropFrontBehavior;
                break;
            case InputMode::DROPCURRENT:
                async_full_behavior = &Input::ASyncFullDropCurrentBehavior;
                break;
            default:
                break;
        }
    }

    bool Flow::SetAsSource(const std::vector<int>& output_slots, FunctionProcess f, const std::string& mark)
    {
        source_start_cond_mtx = std::make_shared<ConditionLockMutex>();
        if (!source_start_cond_mtx) {
            return false;
        }
        SlotMap sm;
        sm.input_slots = std::vector<int>{0};
        sm.output_slots = output_slots;
        sm.process = f;
        sm.thread_model = Model::SYNC;
        sm.mode_when_full = InputMode::DROPFRONT;
        if (!InstallSlotMap(sm, mark, 0)) {
            LOG("Fail to InstallSlotMap, read %s\n", mark.c_str());
            return false;
        }
        return true;
    }

    bool Flow::InstallSlotMap(SlotMap& map, const std::string& mark, int exp_process_time)
    {
        LOGD("%s, thread_model=%d, mode_when_full=%d\n", mark.c_str(), map.thread_model, map.mode_when_full);
        // parameters validity check
        auto& in_slots = map.input_slots;
        if (in_slots.size() > 1 && map.thread_model == Model::SYNC) {
            LOG("More than 1 input to flow, can not set sync input\n");
            return false;
        }
        if (!check_slots(in_slots, "input")) {
            return false;
        }
        bool ret = true;
        for (int i : in_slots) {
            if (i >= (int)v_input.size()) {
                continue;
            }
            if (v_input[i].valid) {
                LOG("input slot %d has been set\n", i);
                ret = false;
            }
        }
        if (!ret) {
            return false;
        }

        auto& out_slots = map.output_slots;
        if (!check_slots(out_slots, "output")) {
            return false;
        }
        for (int i : out_slots) {
            if (i >= (int)downflowmap.size()) {
                continue;
            }
            if (downflowmap[i].valid) {
                LOG("output slot %d has been set\n", i);
                ret = false;
            }
        }
        if (!ret) {
            return false;
        }

        auto c = std::make_shared<FlowCoroutine>(this, map.thread_model, map.process, map.interval);
        if (!c) {
            errno = ENOMEM;
            return false;
        }
        c->Bind(in_slots, out_slots);
        coroutines.push_back(c);
        if (!in_slots.empty()) {
            int max_idx = in_slots[in_slots.size() - 1];
            if ((int)v_input.size() <= max_idx) {
                v_input.resize(max_idx + 1);
            }
            for (size_t i = 0; i < in_slots.size(); i++) {
                v_input[in_slots[i]].Init(
                    this, map.thread_model, (map.thread_model == Model::ASYNCCOMMON) ? map.input_maxcachenum[i] : 0,
                    map.mode_when_full,
                    (map.thread_model == Model::ASYNCCOMMON && map.fetch_block.size() > i) ? map.fetch_block[i] : true,
                    c);
                input_slot_num++;
            }
        }
        if (!out_slots.empty()) {
            int max_idx = out_slots[out_slots.size() - 1];
            if ((int)downflowmap.size() <= max_idx) {
                downflowmap.resize(max_idx + 1);
            }
            for (size_t i = 0; i < out_slots.size(); i++) {
                downflowmap[out_slots[i]].Init(map.thread_model,
                                               map.hold_input.size() > i ? map.hold_input[i] : HoldInputMode::NONE);
                out_slot_num++;
            }
        }

        c->SetMarkName(mark);
        c->SetExpectProcessTime(exp_process_time);
        c->Start();
        return true;
    }

    void Flow::FlowMap::AddFlow(std::shared_ptr<Flow> flow, int index)
    {
        AutoLockMutex _lg(list_mtx);
        auto i = std::find(flows.begin(), flows.end(), flow);
        if (i != flows.end()) {
            LOG("repeatedly add, update index\n");
            i->index_of_in = index;
            return;
        }
        // TODO: sort by sync type in downflow
        flows.emplace_back(flow, index);
    }

    void Flow::FlowMap::RemoveFlow(std::shared_ptr<Flow> flow)
    {
        AutoLockMutex _lg(list_mtx);
        flows.remove_if([&flow](FlowInputMap& fm) {
            return fm == flow;
        });
    }

    bool Flow::AddDownFlow(std::shared_ptr<Flow> down, int out_slot_index, int in_slot_index_of_down)
    {
        if (out_slot_num <= 0 || (int)downflowmap.size() != out_slot_num) {
            LOG("Uncompleted or final flow\n");
            return false;
        }
        if (out_slot_index >= (int)downflowmap.size()) {
            LOG("output slot index exceed max\n");
            return false;
        }
        if (this == down.get()) {
            LOG("can not set self loop flow\n");
            return false;
        }
        downflowmap[out_slot_index].AddFlow(down, in_slot_index_of_down);
        if (source_start_cond_mtx) {
            source_start_cond_mtx->lock();
            down_flow_num++;
            source_start_cond_mtx->notify();
            source_start_cond_mtx->unlock();
        }
        return true;
    }

    void Flow::RemoveDownFlow(std::shared_ptr<Flow> down)
    {
        if (out_slot_num <= 0 || (int)downflowmap.size() != out_slot_num) {
            return;
        }
        // if (down->down_flow_num > 0)
        //   LOG("the removing flow has down flows, remove them first\n");
        for (auto& dm : downflowmap) {
            if (!dm.valid) {
                continue;
            }
            dm.RemoveFlow(down);
            if (source_start_cond_mtx) {
                source_start_cond_mtx->lock();
                down_flow_num--;
                source_start_cond_mtx->notify();
                source_start_cond_mtx->unlock();
            }
        }
    }

    void Flow::SendInput(std::shared_ptr<MediaBuffer>& input, int in_slot_index)
    {
        if (in_slot_index < 0 || in_slot_index >= input_slot_num) {
            errno = EINVAL;
            LOG("ERROR: Input slot[%d] is vaild!\n", in_slot_index);
            return;
        }
        if (enable) {
            auto& in = v_input[in_slot_index];
            CALL_MEMBER_FN(in, in.send_input_behavior)(input);
        }
    }

    bool Flow::SetOutput(const std::shared_ptr<MediaBuffer>& output, int out_slot_index)
    {
        if (out_slot_index < 0 || out_slot_index >= out_slot_num) {
            errno = EINVAL;
            LOG("ERROR: Output slot[%d] is vaild!\n", out_slot_index);
            return false;
        }

        if (out_callback_ && output) {
            out_callback_(out_handler_, output);
        }

        if (enable) {
            auto& out = downflowmap[out_slot_index];
            CALL_MEMBER_FN(out, out.set_output_behavior)(output);
            return true;
        }
        return false;
    }

    bool Flow::ParseWrapFlowParams(const char* param, std::map<std::string, std::string>& flow_params,
                                   std::list<std::string>& sub_param_list)
    {
        sub_param_list = ParseFlowParamToList(param);
        if (sub_param_list.empty()) {
            return false;
        }
        if (!parse_media_param_map(sub_param_list.front().c_str(), flow_params)) {
            return false;
        }
        sub_param_list.pop_front();
        if (flow_params[KEY_NAME].empty()) {
            LOG("missing key name\n");
            return false;
        }
        return true;
    }

    void Flow::RegisterEventHandler(std::shared_ptr<Flow> flow, EventHook proc)
    {
        event_handler_.reset(new EventHandler());
        if (event_handler_) {
            event_handler_->RegisterEventHook(flow, proc);
        }
    }

    void Flow::UnRegisterEventHandler()
    {
        if (event_handler_) {
            event_handler_->UnRegisterEventHook();
            event_handler_.reset(nullptr);
        }
    }

    void Flow::NotifyToEventHandler(EventParamPtr param, int type)
    {
        if (event_handler_) {
            MessagePtr msg = std::make_shared<EventMessage>(this, param, type);
            event_handler_->NotifyToEventHandler(msg);
            event_handler_->SignalEventHook();
        }
    }

    void Flow::NotifyToEventHandler(int id, int type)
    {
        if (event_handler_) {
            EventParamPtr event_param = std::make_shared<EventParam>(id, 0);
            MessagePtr msg = std::make_shared<EventMessage>(this, event_param, type);
            event_handler_->NotifyToEventHandler(msg);
            event_handler_->SignalEventHook();
        }
    }

    void Flow::EventHookWait()
    {
        if (event_handler_) {
            event_handler_->EventHookWait();
        }
    }

    MessagePtr Flow::GetEventMessage()
    {
        if (event_handler_) {
            return event_handler_->GetEventMessage();
        }
        return nullptr;
    }

    EventParamPtr Flow::GetEventParam(MessagePtr msg)
    {
        if (msg) {
            return msg->GetEventParam();
        }
        return nullptr;
    }

    void Flow::Input::SyncSendInputBehavior(std::shared_ptr<MediaBuffer>& input)
    {
        cached_buffer = input;
        coroutine->RunOnce();
        cached_buffer.reset();
    }

    void Flow::Input::ASyncSendInputCommonBehavior(std::shared_ptr<MediaBuffer>& input)
    {
        mtx.lock();
        if (max_cache_num > 0 && max_cache_num <= (int)cached_buffers.size()) {
            bool ret = (this->*async_full_behavior)(flow->enable);
            if (!ret) {
                mtx.unlock();
                return;
            }
        }
        cached_buffers.push_back(input);
        mtx.unlock();
        AutoLockMutex _alm(flow->cond_mtx);
        flow->cond_mtx.notify();
    }

    void Flow::Input::ASyncSendInputAtomicBehavior(std::shared_ptr<MediaBuffer>& input)
    {
        AutoLockMutex _alm(spin_mtx);
        cached_buffer = input;
    }

    bool Flow::Input::ASyncFullBlockingBehavior(volatile bool& pred)
    {
#ifndef NDEBUG
        AutoDuration ad;
#endif
        do {
            mtx.unlock();
            msleep(5);
            if (max_cache_num > (int)cached_buffers.size()) {
                mtx.lock();
                break;
            }
            mtx.lock();
        } while (pred);
#ifndef NDEBUG
        if (ad.Get() > 100000 /*ms*/)
            LOG("WARN: Flow[%s]: Input[block mode]: block too long(%.2fms) > 5ms\n",
                flow ? flow->GetFlowTag() : "Name is null", ad.Get() / 1000.0);
#endif
        return pred;
    }

    bool Flow::Input::ASyncFullDropFrontBehavior(volatile bool& pred _UNUSED)
    {
        LOG("WARN: Flow[%s]: Input: drop front buffer!\n", flow ? flow->GetFlowTag() : "Name is null");
        cached_buffers.pop_front();
        return true;
    }

    bool Flow::Input::ASyncFullDropCurrentBehavior(volatile bool& pred _UNUSED)
    {
        LOG("WARN: Flow[%s]: Input: drop current buffer!\n", flow ? flow->GetFlowTag() : "Name Is Null");
        return false;
    }

    std::string gen_datatype_rule(std::map<std::string, std::string>& params)
    {
        std::string rule;
        std::string value;
        CHECK_EMPTY_SETERRNO_RETURN(, value, params, KEY_INPUTDATATYPE, , "");
        PARAM_STRING_APPEND(rule, KEY_INPUTDATATYPE, value);
        CHECK_EMPTY_SETERRNO_RETURN(, value, params, KEY_OUTPUTDATATYPE, , "");
        PARAM_STRING_APPEND(rule, KEY_OUTPUTDATATYPE, value);
        return std::move(rule);
    }

    Model GetModelByString(const std::string& model)
    {
        static std::map<std::string, Model> model_map = {
            {KEY_ASYNCCOMMON, Model::ASYNCCOMMON}, {KEY_ASYNCATOMIC, Model::ASYNCATOMIC}, {KEY_SYNC, Model::SYNC}};
        auto it = model_map.find(model);
        if (it != model_map.end()) {
            return it->second;
        }
        return Model::NONE;
    }

    InputMode GetInputModelByString(const std::string& in_model)
    {
        static std::map<std::string, InputMode> in_model_map = {{KEY_BLOCKING, InputMode::BLOCKING},
                                                                {KEY_DROPFRONT, InputMode::DROPFRONT},
                                                                {KEY_DROPCURRENT, InputMode::DROPCURRENT}};
        auto it = in_model_map.find(in_model);
        if (it != in_model_map.end()) {
            return it->second;
        }
        return InputMode::NONE;
    }

    void ParseParamToSlotMap(std::map<std::string, std::string>& params, SlotMap& sm, int& input_maxcachenum)
    {
        float fps = 0.0f;
        std::string& fps_str = params[KEY_FPS];
        if (!fps_str.empty()) {
            fps = std::stof(fps_str);
            if (fps > 0.0f) {
                sm.interval = 1000.0f / fps;
            }
        }
        sm.thread_model = GetModelByString(params[KEK_THREAD_SYNC_MODEL]);
        sm.mode_when_full = GetInputModelByString(params[KEK_INPUT_MODEL]);
        std::string& cache_num_str = params[KEY_INPUT_CACHE_NUM];
        int cache_num = -1;
        if (!cache_num_str.empty()) {
            cache_num = std::stoi(cache_num_str);
            if (cache_num <= 0) {
                LOG("warning, input cache num = %d\n", cache_num);
            }
            input_maxcachenum = cache_num;
        }
    }

    size_t FlowOutputHoldInput(std::shared_ptr<MediaBuffer>& out_buffer, const MediaBufferVector& input_vector)
    {
        assert(out_buffer);
        size_t i = 0;
        for (; i < input_vector.size(); i++) {
            out_buffer->SetRelatedSPtr(input_vector[i], i);
        }
        return i;
    }

    size_t FlowOutputInheritFromInput(std::shared_ptr<MediaBuffer>& out_buffer, const MediaBufferVector& input_vector)
    {
        assert(out_buffer);
        size_t ret = 0;
        auto& vec = out_buffer->GetRelatedSPtrs();
        for (size_t i = 0; i < input_vector.size(); i++) {
            auto& input_vec = input_vector[i]->GetRelatedSPtrs();
            ret += input_vec.size();
            vec.insert(vec.end(), input_vec.begin(), input_vec.end());
        }
        return ret;
    }

    std::string JoinFlowParam(const std::string& flow_param, size_t num_elem, ...)
    {
        std::string ret;
        ret.append(flow_param);
        va_list va;
        va_start(va, num_elem);
        while (num_elem--) {
            const std::string& elem_param = va_arg(va, const std::string);
            ret.append(1, FLOW_PARAM_SEPARATE_CHAR).append(elem_param);
        }
        va_end(va);
        return std::move(ret);
    }

    std::list<std::string> ParseFlowParamToList(const char* param)
    {
        std::list<std::string> separate_list;
        if (!parse_media_param_list(param, separate_list, FLOW_PARAM_SEPARATE_CHAR) || separate_list.size() < 2) {
            separate_list.clear();
        }
        return std::move(separate_list);
    }

} // namespace easymedia