musicdsp/audio-river
1
0
Fork 0
Code Issues Pull Requests Releases Wiki Activity

[DEV] change process pointer to direct

Browse Source
This commit is contained in:
Edouard DUPIN 2015-02-09 21:44:32 +01:00
parent 8896c0b5de
commit d3e2aa1d43
3 changed files with 60 additions and 48 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

70
river/Interface.cpp
View File

@ -36,39 +36,38 @@ bool river::Interface::init(const std::string& _name,
m_freq = _freq;
m_map = _map;
m_format = _format;
m_process = std::make_shared<drain::Process>();
m_volume = 0.0f;
// register interface to be notify from the volume change.
m_node->registerAsRemote(shared_from_this());
// Create convertion interface
if (m_node->isInput() == true) {
m_process->setInputConfig(m_node->getInterfaceFormat());
m_process.setInputConfig(m_node->getInterfaceFormat());
// add all time the volume stage :
std::shared_ptr<drain::Volume> algo = drain::Volume::create();
//algo->setInputFormat(m_node->getInterfaceFormat());
algo->setName("volume");
m_process->pushBack(algo);
m_process.pushBack(algo);
RIVER_INFO("add basic volume stage (1)");
std::shared_ptr<drain::VolumeElement> tmpVolume = m_node->getVolume();
if (tmpVolume != nullptr) {
RIVER_INFO(" add volume for node");
algo->addVolumeStage(tmpVolume);
}
m_process->setOutputConfig(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process.setOutputConfig(drain::IOFormatInterface(m_map, m_format, m_freq));
} else {
m_process->setInputConfig(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process.setInputConfig(drain::IOFormatInterface(m_map, m_format, m_freq));
// add all time the volume stage :
std::shared_ptr<drain::Volume> algo = drain::Volume::create();
//algo->setOutputFormat(m_node->getInterfaceFormat());
algo->setName("volume");
m_process->pushBack(algo);
m_process.pushBack(algo);
RIVER_INFO("add basic volume stage (2)");
std::shared_ptr<drain::VolumeElement> tmpVolume = m_node->getVolume();
if (tmpVolume != nullptr) {
RIVER_INFO(" add volume for node");
algo->addVolumeStage(tmpVolume);
}
m_process->setOutputConfig(m_node->getInterfaceFormat());
m_process.setOutputConfig(m_node->getInterfaceFormat());
}
return true;
}
@ -87,7 +86,6 @@ river::Interface::~Interface() {
//stop(true, true);
std::unique_lock<std::recursive_mutex> lock(m_mutex);
//m_node->interfaceRemove(shared_from_this());
m_process.reset();
}
/*
bool river::Interface::hasEndPoint() {
@ -96,47 +94,47 @@ bool river::Interface::hasEndPoint() {
*/
void river::Interface::setReadwrite() {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->removeAlgoDynamic();
if (m_process->hasType<drain::EndPoint>() ) {
m_process.removeAlgoDynamic();
if (m_process.hasType<drain::EndPoint>() ) {
RIVER_ERROR("Endpoint is already present ==> can not change");
return;
}
if (m_node->isInput() == true) {
m_process->removeIfLast<drain::EndPoint>();
m_process.removeIfLast<drain::EndPoint>();
std::shared_ptr<drain::EndPointRead> algo = drain::EndPointRead::create();
//algo->setOutputFormat(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process->pushBack(algo);
m_process.pushBack(algo);
} else {
m_process->removeIfFirst<drain::EndPoint>();
m_process.removeIfFirst<drain::EndPoint>();
std::shared_ptr<drain::EndPointWrite> algo = drain::EndPointWrite::create();
//algo->setInputFormat(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process->pushFront(algo);
m_process.pushFront(algo);
}
}
void river::Interface::setOutputCallback(size_t _chunkSize, drain::needDataFunction _function) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->removeAlgoDynamic();
m_process->removeIfFirst<drain::EndPoint>();
m_process.removeAlgoDynamic();
m_process.removeIfFirst<drain::EndPoint>();
std::shared_ptr<drain::Algo> algo = drain::EndPointCallback::create(_function);
RIVER_INFO("set property: " << m_map << " " << m_format << " " << m_freq);
//algo->setInputFormat(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process->pushFront(algo);
m_process.pushFront(algo);
}
void river::Interface::setInputCallback(size_t _chunkSize, drain::haveNewDataFunction _function) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->removeAlgoDynamic();
m_process->removeIfLast<drain::EndPoint>();
m_process.removeAlgoDynamic();
m_process.removeIfLast<drain::EndPoint>();
std::shared_ptr<drain::Algo> algo = drain::EndPointCallback::create(_function);
//algo->setOutputFormat(drain::IOFormatInterface(m_map, m_format, m_freq));
m_process->pushBack(algo);
m_process.pushBack(algo);
}
void river::Interface::setWriteCallback(drain::needDataFunctionWrite _function) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->removeAlgoDynamic();
std::shared_ptr<drain::EndPointWrite> algo = m_process->get<drain::EndPointWrite>(0);
m_process.removeAlgoDynamic();
std::shared_ptr<drain::EndPointWrite> algo = m_process.get<drain::EndPointWrite>(0);
if (algo == nullptr) {
return;
}
@ -146,7 +144,7 @@ void river::Interface::setWriteCallback(drain::needDataFunctionWrite _function)
void river::Interface::start(const std::chrono::system_clock::time_point& _time) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
RIVER_DEBUG("start [BEGIN]");
m_process->updateInterAlgo();
m_process.updateInterAlgo();
m_node->interfaceAdd(shared_from_this());
RIVER_DEBUG("start [ END ]");
}
@ -173,7 +171,7 @@ bool river::Interface::setParameter(const std::string& _filter, const std::strin
RIVER_ERROR("Interface is not allowed to modify '" << _parameter << "' Volume just allowed to modify 'FLOW' volume");
return false;
}
std::shared_ptr<drain::Algo> algo = m_process->get<drain::Algo>(_filter);
std::shared_ptr<drain::Algo> algo = m_process.get<drain::Algo>(_filter);
if (algo == nullptr) {
RIVER_ERROR("setParameter(" << _filter << ") ==> no filter named like this ...");
return false;
@ -185,7 +183,7 @@ bool river::Interface::setParameter(const std::string& _filter, const std::strin
std::string river::Interface::getParameter(const std::string& _filter, const std::string& _parameter) const {
RIVER_DEBUG("getParameter [BEGIN] : '" << _filter << "':'" << _parameter << "'");
std::string out;
std::shared_ptr<drain::Algo> algo = m_process->get<drain::Algo>(_filter);
std::shared_ptr<const drain::Algo> algo = m_process.get<const drain::Algo>(_filter);
if (algo == nullptr) {
RIVER_ERROR("setParameter(" << _filter << ") ==> no filter named like this ...");
return "[ERROR]";
@ -197,7 +195,7 @@ std::string river::Interface::getParameter(const std::string& _filter, const std
std::string river::Interface::getParameterProperty(const std::string& _filter, const std::string& _parameter) const {
RIVER_DEBUG("getParameterProperty [BEGIN] : '" << _filter << "':'" << _parameter << "'");
std::string out;
std::shared_ptr<drain::Algo> algo = m_process->get<drain::Algo>(_filter);
std::shared_ptr<const drain::Algo> algo = m_process.get<const drain::Algo>(_filter);
if (algo == nullptr) {
RIVER_ERROR("setParameter(" << _filter << ") ==> no filter named like this ...");
return "[ERROR]";
@ -209,8 +207,8 @@ std::string river::Interface::getParameterProperty(const std::string& _filter, c
void river::Interface::write(const void* _value, size_t _nbChunk) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->updateInterAlgo();
std::shared_ptr<drain::EndPointWrite> algo = m_process->get<drain::EndPointWrite>(0);
m_process.updateInterAlgo();
std::shared_ptr<drain::EndPointWrite> algo = m_process.get<drain::EndPointWrite>(0);
if (algo == nullptr) {
return;
}
@ -244,7 +242,7 @@ std::vector<int16_t> river::Interface::read(size_t _nbChunk) {
void river::Interface::read(void* _value, size_t _nbChunk) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->updateInterAlgo();
m_process.updateInterAlgo();
// TODO :...
}
@ -257,21 +255,21 @@ size_t river::Interface::size() const {
void river::Interface::setBufferSize(size_t _nbChunk) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->updateInterAlgo();
m_process.updateInterAlgo();
// TODO :...
}
void river::Interface::setBufferSize(const std::chrono::duration<int64_t, std::micro>& _time) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->updateInterAlgo();
m_process.updateInterAlgo();
// TODO :...
}
void river::Interface::clearInternalBuffer() {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
m_process->updateInterAlgo();
m_process.updateInterAlgo();
// TODO :...
}
@ -286,7 +284,7 @@ std::chrono::system_clock::time_point river::Interface::getCurrentTime() const {
void river::Interface::addVolumeGroup(const std::string& _name) {
std::unique_lock<std::recursive_mutex> lock(m_mutex);
RIVER_DEBUG("addVolumeGroup(" << _name << ")");
std::shared_ptr<drain::Volume> algo = m_process->get<drain::Volume>("volume");
std::shared_ptr<drain::Volume> algo = m_process.get<drain::Volume>("volume");
if (algo == nullptr) {
RIVER_ERROR("addVolumeGroup(" << _name << ") ==> no volume stage ... can not add it ...");
return;
@ -303,17 +301,17 @@ void river::Interface::addVolumeGroup(const std::string& _name) {
void river::Interface::systemNewInputData(std::chrono::system_clock::time_point _time, void* _data, size_t _nbChunk) {
std::unique_lock<std::recursive_mutex> lockProcess(m_mutex);
m_process->push(_time, _data, _nbChunk);
m_process.push(_time, _data, _nbChunk);
}
void river::Interface::systemNeedOutputData(std::chrono::system_clock::time_point _time, void* _data, size_t _nbChunk, size_t _chunkSize) {
std::unique_lock<std::recursive_mutex> lockProcess(m_mutex);
m_process->pull(_time, _data, _nbChunk, _chunkSize);
m_process.pull(_time, _data, _nbChunk, _chunkSize);
}
void river::Interface::systemVolumeChange() {
std::unique_lock<std::recursive_mutex> lockProcess(m_mutex);
std::shared_ptr<drain::Volume> algo = m_process->get<drain::Volume>("volume");
std::shared_ptr<drain::Volume> algo = m_process.get<drain::Volume>("volume");
if (algo == nullptr) {
return;
}

2
river/Interface.h
View File

@ -34,7 +34,7 @@ namespace river {
float m_freq;
std::vector<audio::channel> m_map;
audio::format m_format;
std::shared_ptr<drain::Process> m_process;
drain::Process m_process;
protected:
std::string m_name;
public:

36
river/io/Node.cpp
View File

@ -26,39 +26,51 @@
#endif
int32_t river::io::Node::rtAudioCallback(void* _outputBuffer,
void* _inputBuffer,
unsigned int _nBufferFrames,
double _streamTime,
airtaudio::status _status) {
void* _inputBuffer,
unsigned int _nBufferFrames,
double _streamTime,
airtaudio::status _status) {
std::unique_lock<std::mutex> lock(m_mutex);
std::chrono::system_clock::time_point ttime = std::chrono::system_clock::time_point();//std::chrono::system_clock::now();
/*
for (int32_t iii=0; iii<400; ++iii) {
RIVER_VERBOSE("dummy=" << uint64_t(dummy[iii]));
}
*/
if (_outputBuffer != nullptr) {
RIVER_VERBOSE("data Output");
RIVER_VERBOSE("data Output size request :" << _nBufferFrames << " [BEGIN] status=" << _status);
std::vector<int32_t> output;
RIVER_VERBOSE("resize=" << _nBufferFrames*m_process.getInputConfig().getMap().size());
output.resize(_nBufferFrames*m_process.getInputConfig().getMap().size(), 0);
const int32_t* outputTmp = nullptr;
std::vector<uint8_t> outputTmp2;
RIVER_VERBOSE("resize=" << sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nBufferFrames);
outputTmp2.resize(sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nBufferFrames, 0);
int32_t id = 1;
for (auto &it : m_list) {
RIVER_VERBOSE(" IO : " << id << "/" << m_list.size() << " pointer : " << uint64_t(&(*it)));
if (it != nullptr) {
RIVER_VERBOSE(" name="<< it->getName());
// clear datas ...
memset(&outputTmp2[0], 0, sizeof(int32_t)*m_process.getInputConfig().getMap().size()*_nBufferFrames);
RIVER_VERBOSE(" IO : " /* << std::distance(m_list.begin(), it)*/ << "/" << m_list.size() << " name="<< it->getName());
RIVER_VERBOSE(" request Data="<< _nBufferFrames);
it->systemNeedOutputData(ttime, &outputTmp2[0], _nBufferFrames, sizeof(int32_t)*m_process.getInputConfig().getMap().size());
RIVER_VERBOSE(" Mix it ...");
outputTmp = reinterpret_cast<const int32_t*>(&outputTmp2[0]);
// Add data to the output tmp buffer :
for (size_t kkk=0; kkk<output.size(); ++kkk) {
output[kkk] += outputTmp[kkk];
}
break;
}
++id;
}
RIVER_VERBOSE(" End stack process data ...");
m_process.processIn(&outputTmp2[0], _nBufferFrames, _outputBuffer, _nBufferFrames);
// TODO : Call feedback ...
RIVER_VERBOSE("data Output size request :" << _nBufferFrames << " [ END ]");
}
if (_inputBuffer != nullptr) {
RIVER_INFO("data Input");
RIVER_VERBOSE("data Input size request :" << _nBufferFrames << " [BEGIN]");
int16_t* inputBuffer = static_cast<int16_t *>(_inputBuffer);
for (size_t iii=0; iii< m_list.size(); ++iii) {
if (m_list[iii] != nullptr) {
@ -66,6 +78,7 @@ int32_t river::io::Node::rtAudioCallback(void* _outputBuffer,
m_list[iii]->systemNewInputData(ttime, inputBuffer, _nBufferFrames);
}
}
RIVER_VERBOSE("data Input size request :" << _nBufferFrames << " [ END ]");
}
return 0;
}
@ -201,7 +214,7 @@ river::io::Node::Node(const std::string& _name, const std::shared_ptr<const ejso
enum airtaudio::error err = airtaudio::error_none;
if (m_isInput == true) {
err = m_adac.openStream(nullptr, &params,
audio::format_int16, hardwareFormat.getFrequency(), &m_rtaudioFrameSize,
hardwareFormat.getFormat(), hardwareFormat.getFrequency(), &m_rtaudioFrameSize,
std::bind(&river::io::Node::rtAudioCallback,
this,
std::placeholders::_1,
@ -212,7 +225,7 @@ river::io::Node::Node(const std::string& _name, const std::shared_ptr<const ejso
);
} else {
err = m_adac.openStream(&params, nullptr,
audio::format_int16, hardwareFormat.getFrequency(), &m_rtaudioFrameSize,
hardwareFormat.getFormat(), hardwareFormat.getFrequency(), &m_rtaudioFrameSize,
std::bind(&river::io::Node::rtAudioCallback,
this,
std::placeholders::_1,
@ -232,6 +245,7 @@ river::io::Node::Node(const std::string& _name, const std::shared_ptr<const ejso
m_process.setOutputConfig(hardwareFormat);
m_process.setInputConfig(interfaceFormat);
}
m_process.updateInterAlgo();
}
river::io::Node::~Node() {