// 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.
|
|
#ifndef EASYMEDIA_FLOW_H_
|
#define EASYMEDIA_FLOW_H_
|
|
#include <stdarg.h>
|
|
#include <deque>
|
#include <thread>
|
#include <type_traits>
|
#include <vector>
|
|
#include "control.h"
|
#include "lock.h"
|
#include "message.h"
|
#include "reflector.h"
|
#include "utils.h"
|
|
namespace easymedia
|
{
|
|
DECLARE_FACTORY(Flow)
|
// usage: REFLECTOR(Flow)::Create<T>(flowname, param)
|
// T must be the final class type exposed to user
|
DECLARE_REFLECTOR(Flow)
|
|
#define DEFINE_FLOW_FACTORY(REAL_PRODUCT, FINAL_EXPOSE_PRODUCT) \
|
DEFINE_MEDIA_CHILD_FACTORY(REAL_PRODUCT, REAL_PRODUCT::GetFlowName(), FINAL_EXPOSE_PRODUCT, Flow) \
|
DEFINE_MEDIA_CHILD_FACTORY_EXTRA(REAL_PRODUCT) \
|
DEFINE_MEDIA_NEW_PRODUCT_BY(REAL_PRODUCT, FINAL_EXPOSE_PRODUCT, GetError() < 0)
|
|
class MediaBuffer;
|
enum class Model
|
{
|
NONE,
|
ASYNCCOMMON,
|
ASYNCATOMIC,
|
SYNC
|
};
|
// PushMode
|
enum class InputMode
|
{
|
NONE,
|
BLOCKING,
|
DROPFRONT,
|
DROPCURRENT
|
};
|
enum class HoldInputMode
|
{
|
NONE,
|
HOLD_INPUT,
|
INHERIT_FORM_INPUT
|
};
|
using MediaBufferVector = std::vector<std::shared_ptr<MediaBuffer>>;
|
// TODO: outputs ret, outslot index, outslot queue model
|
using FunctionProcess = std::add_pointer<bool(Flow* f, MediaBufferVector& input_vector)>::type;
|
template <int in_index, int out_index> bool void_transaction(Flow* f, MediaBufferVector& input_vector);
|
using LinkVideoHandler = std::add_pointer<void(unsigned char* buffer, unsigned int buffer_size,
|
int64_t present_time, int nat_type)>::type;
|
using LinkAudioHandler =
|
std::add_pointer<void(unsigned char* buffer, unsigned int buffer_size, int64_t present_time)>::type;
|
using LinkCaptureHandler =
|
std::add_pointer<void(unsigned char* buffer, unsigned int buffer_size, int type, uint32_t sequence)>::type;
|
using PlayVideoHandler = std::add_pointer<void(Flow* f)>::type;
|
using PlayAudioHandler = std::add_pointer<void(Flow* f)>::type;
|
using CallBackHandler = std::add_pointer<void>::type;
|
using UserCallBack = std::add_pointer<void(void* handler, int type, void* ptr, int size)>::type;
|
using OutputCallBack = std::add_pointer<void(void* handler, std::shared_ptr<MediaBuffer> mb)>::type;
|
using EventCallBack = std::add_pointer<void(void* handler, void* data)>::type;
|
|
class _API SlotMap
|
{
|
public:
|
SlotMap() : thread_model(Model::SYNC), mode_when_full(InputMode::DROPFRONT), process(nullptr), interval(16.66f)
|
{
|
}
|
std::vector<int> input_slots;
|
Model thread_model;
|
InputMode mode_when_full;
|
std::vector<bool> fetch_block; // if ASYNCCOMMON
|
std::vector<int> input_maxcachenum;
|
std::vector<int> output_slots;
|
// std::vector<DataSetModel> output_ds_model;
|
std::vector<HoldInputMode> hold_input;
|
FunctionProcess process;
|
float interval;
|
};
|
|
class FlowCoroutine;
|
class _API Flow
|
{
|
public:
|
// We may need a flow which can be sync and async.
|
// This make a side effect that sync flow contains superfluous variables
|
// designed for async implementation.
|
Flow();
|
virtual ~Flow();
|
static const char* GetFlowName()
|
{
|
return nullptr;
|
}
|
// The GetFlowName interface is occupied by the reflector,
|
// so GetFlowTag is used to distinguish Flow.
|
const char* GetFlowTag()
|
{
|
return flow_tag.c_str();
|
}
|
void SetFlowTag(std::string tag)
|
{
|
flow_tag = tag;
|
}
|
|
// TODO: Right now out_slot_index and in_slot_index is decided by exact
|
// subclass, automatically get these value or ignore them in future.
|
bool AddDownFlow(std::shared_ptr<Flow> down, int out_slot_index, int in_slot_index_of_down);
|
void RemoveDownFlow(std::shared_ptr<Flow> down);
|
|
void SendInput(std::shared_ptr<MediaBuffer>& input, int in_slot_index);
|
void SetDisable()
|
{
|
enable = false;
|
}
|
|
// The Control must be called in the same thread to that create flow
|
virtual int Control(unsigned long int request _UNUSED, ...)
|
{
|
return -1;
|
}
|
virtual int SubControl(unsigned long int request, void* arg, int size = 0)
|
{
|
SubRequest subreq = {request, size, arg};
|
return Control(S_SUB_REQUEST, &subreq);
|
}
|
|
// get input size for this flow
|
virtual int GetInputSize()
|
{
|
return 0;
|
}
|
|
// The global event hander is the same thread to the born thread of this
|
// object.
|
void RegisterEventHandler(std::shared_ptr<Flow> flow, EventHook proc);
|
void UnRegisterEventHandler();
|
void EventHookWait();
|
void NotifyToEventHandler(EventParamPtr param, int type = MESSAGE_TYPE_FIFO);
|
void NotifyToEventHandler(int id, int type = MESSAGE_TYPE_FIFO);
|
MessagePtr GetEventMessage();
|
EventParamPtr GetEventParam(MessagePtr msg);
|
|
// Add Link hander For app Link
|
void SetVideoHandler(LinkVideoHandler hander)
|
{
|
link_video_handler_ = hander;
|
}
|
LinkVideoHandler GetVideoHandler()
|
{
|
return link_video_handler_;
|
}
|
void SetAudioHandler(LinkAudioHandler hander)
|
{
|
link_audio_handler_ = hander;
|
}
|
LinkAudioHandler GetAudioHandler()
|
{
|
return link_audio_handler_;
|
}
|
void SetCaptureHandler(LinkCaptureHandler hander)
|
{
|
link_capture_handler_ = hander;
|
}
|
LinkCaptureHandler GetCaptureHandler()
|
{
|
return link_capture_handler_;
|
}
|
|
// Add hander For rtsp flow
|
void SetPlayVideoHandler(PlayVideoHandler handler)
|
{
|
play_video_handler_ = handler;
|
}
|
PlayVideoHandler GetPlayVideoHandler()
|
{
|
return play_video_handler_;
|
}
|
void SetPlayAudioHandler(PlayAudioHandler handler)
|
{
|
play_audio_handler_ = handler;
|
}
|
PlayAudioHandler GetPlayAudioHandler()
|
{
|
return play_audio_handler_;
|
}
|
|
// Add common hander for user
|
void SetUserCallBack(CallBackHandler handler, UserCallBack callback)
|
{
|
user_handler_ = handler;
|
user_callback_ = callback;
|
}
|
void SetOutputCallBack(CallBackHandler handler, OutputCallBack callback)
|
{
|
out_handler_ = handler;
|
out_callback_ = callback;
|
}
|
void SetEventCallBack(CallBackHandler handler, EventCallBack callback)
|
{
|
event_handler2_ = handler;
|
event_callback_ = callback;
|
}
|
CallBackHandler GetUserHandler()
|
{
|
return user_handler_;
|
}
|
UserCallBack GetUserCallBack()
|
{
|
return user_callback_;
|
}
|
|
// Control the number of executions of threads inside Flow
|
// _run_times: -1, Endless loop; 0, skip process; > 0, do process cnt.
|
int SetRunTimes(int _run_times);
|
int GetRunTimesRemaining();
|
|
bool IsAllBuffEmpty();
|
void DumpBase(std::string& dump_info);
|
virtual void Dump(std::string& dump_info)
|
{
|
DumpBase(dump_info);
|
}
|
|
void StartStream();
|
|
protected:
|
class FlowInputMap
|
{
|
public:
|
FlowInputMap(std::shared_ptr<Flow>& f, int i) : flow(f), index_of_in(i)
|
{
|
}
|
std::shared_ptr<Flow> flow; // weak_ptr?
|
int index_of_in;
|
bool operator==(const std::shared_ptr<easymedia::Flow> f)
|
{
|
return flow == f;
|
}
|
};
|
class FlowMap
|
{
|
private:
|
void SetOutputBehavior(const std::shared_ptr<MediaBuffer>& output);
|
void SetOutputToQueueBehavior(const std::shared_ptr<MediaBuffer>& output);
|
|
public:
|
FlowMap() : valid(false), hold_input(HoldInputMode::NONE)
|
{
|
assert(list_mtx.valid);
|
}
|
FlowMap(FlowMap&&);
|
void Init(Model m, HoldInputMode hold_in);
|
bool valid;
|
HoldInputMode hold_input;
|
// down flow
|
void AddFlow(std::shared_ptr<Flow> flow, int index);
|
void RemoveFlow(std::shared_ptr<Flow> flow);
|
std::list<FlowInputMap> flows;
|
ReadWriteLockMutex list_mtx;
|
std::deque<std::shared_ptr<MediaBuffer>> cached_buffers; // never drop
|
std::shared_ptr<MediaBuffer> cached_buffer;
|
decltype(&FlowMap::SetOutputBehavior) set_output_behavior;
|
};
|
class Input
|
{
|
private:
|
void SyncSendInputBehavior(std::shared_ptr<MediaBuffer>& input);
|
void ASyncSendInputCommonBehavior(std::shared_ptr<MediaBuffer>& input);
|
void ASyncSendInputAtomicBehavior(std::shared_ptr<MediaBuffer>& input);
|
// behavior when input list exceed max_cache_num
|
bool ASyncFullBlockingBehavior(volatile bool& pred);
|
bool ASyncFullDropFrontBehavior(volatile bool& pred);
|
bool ASyncFullDropCurrentBehavior(volatile bool& pred);
|
|
public:
|
Input() : valid(false), flow(nullptr), fetch_block(true)
|
{
|
}
|
Input(Input&&);
|
void Init(Flow* f, Model m, int mcn, InputMode im, bool f_block, std::shared_ptr<FlowCoroutine> fc);
|
bool valid;
|
Flow* flow;
|
Model thread_model;
|
bool fetch_block;
|
std::deque<std::shared_ptr<MediaBuffer>> cached_buffers;
|
ConditionLockMutex mtx;
|
int max_cache_num;
|
InputMode mode_when_full;
|
std::shared_ptr<MediaBuffer> cached_buffer;
|
SpinLockMutex spin_mtx;
|
decltype(&Input::SyncSendInputBehavior) send_input_behavior;
|
decltype(&Input::ASyncFullBlockingBehavior) async_full_behavior;
|
std::shared_ptr<FlowCoroutine> coroutine;
|
};
|
|
// Can not change the following values after initialize,
|
// except AddFlow and RemoveFlow.
|
int out_slot_num;
|
std::vector<FlowMap> downflowmap;
|
int input_slot_num;
|
std::vector<Input> v_input;
|
std::list<std::shared_ptr<FlowCoroutine>> coroutines;
|
std::shared_ptr<ConditionLockMutex> source_start_cond_mtx;
|
|
int down_flow_num;
|
bool waite_down_flow;
|
|
// source flow
|
bool SetAsSource(const std::vector<int>& output_slots, FunctionProcess f, const std::string& mark);
|
bool InstallSlotMap(SlotMap& map, const std::string& mark, int exp_process_time);
|
bool SetOutput(const std::shared_ptr<MediaBuffer>& output, int out_slot_index);
|
bool ParseWrapFlowParams(const char* param, std::map<std::string, std::string>& flow_params,
|
std::list<std::string>& sub_param_list);
|
// As sub threads may call the variable of child class,
|
// we should define this for child class when it deconstruct.
|
void StopAllThread();
|
bool IsEnable()
|
{
|
return enable;
|
}
|
|
template <int in_index, int out_index> friend bool void_transaction(Flow* f, MediaBufferVector& input_vector)
|
{
|
return f->SetOutput(input_vector[in_index], out_index);
|
}
|
static const FunctionProcess void_transaction00;
|
|
CallBackHandler event_handler2_;
|
EventCallBack event_callback_;
|
|
private:
|
volatile bool enable;
|
volatile bool quit;
|
ConditionLockMutex cond_mtx;
|
|
// event handler
|
std::unique_ptr<EventHandler> event_handler_;
|
|
friend class FlowCoroutine;
|
|
LinkVideoHandler link_video_handler_;
|
LinkAudioHandler link_audio_handler_;
|
LinkCaptureHandler link_capture_handler_;
|
|
PlayVideoHandler play_video_handler_;
|
PlayAudioHandler play_audio_handler_;
|
|
CallBackHandler user_handler_;
|
UserCallBack user_callback_;
|
|
CallBackHandler out_handler_;
|
OutputCallBack out_callback_;
|
|
// FlowTag is used to distinguish Flow.
|
std::string flow_tag;
|
|
// Control the number of executions of threads inside Flow
|
int run_times;
|
|
DEFINE_ERR_GETSET()
|
DECLARE_PART_FINAL_EXPOSE_PRODUCT(Flow)
|
};
|
|
std::string gen_datatype_rule(std::map<std::string, std::string>& params);
|
Model GetModelByString(const std::string& model);
|
InputMode GetInputModelByString(const std::string& in_model);
|
_API void ParseParamToSlotMap(std::map<std::string, std::string>& params, SlotMap& sm, int& input_maxcachenum);
|
size_t FlowOutputHoldInput(std::shared_ptr<MediaBuffer>& out_buffer, const MediaBufferVector& input_vector);
|
size_t FlowOutputInheritFromInput(std::shared_ptr<MediaBuffer>& out_buffer, const MediaBufferVector& input_vector);
|
|
// the separator of flow params and flow core element params
|
#define FLOW_PARAM_SEPARATE_CHAR ' '
|
_API std::string JoinFlowParam(const std::string& flow_param, size_t num_elem, ...);
|
_API std::list<std::string> ParseFlowParamToList(const char* param);
|
|
} // namespace easymedia
|
|
#endif // #ifndef EASYMEDIA_FLOW_H_
|
