350 lines
8.5 KiB
C++
350 lines
8.5 KiB
C++
/*
|
|
*******************************************************************************
|
|
* INCLUDE
|
|
*******************************************************************************
|
|
*/
|
|
#include "Component.h"
|
|
#include "System.h"
|
|
#include "util/Math.h"
|
|
#include "Logger.h"
|
|
#include <unistd.h>
|
|
|
|
namespace Video
|
|
{
|
|
namespace Core
|
|
{
|
|
/*
|
|
*******************************************************************************
|
|
* CLASS DEFINE
|
|
*******************************************************************************
|
|
*/
|
|
bool Component::Collection::add(Ptr pObject)
|
|
{
|
|
Poco::FastMutex::ScopedLock lock(_lock);
|
|
auto iter = std::find(_list.begin(), _list.end(), pObject);
|
|
if (iter == _list.end())
|
|
{
|
|
_list.push_back(pObject);
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
bool Component::Collection::remove(Ptr pObject)
|
|
{
|
|
Poco::FastMutex::ScopedLock lock(_lock);
|
|
auto iter = std::find(_list.begin(), _list.end(), pObject);
|
|
if (iter != _list.end())
|
|
{
|
|
_list.remove(pObject);
|
|
return true;
|
|
}
|
|
|
|
return false;
|
|
}
|
|
|
|
void Component::Collection::foreach (std::function<void(Ptr)> func)
|
|
{
|
|
std::vector<Ptr> v;
|
|
{
|
|
Poco::FastMutex::ScopedLock lock(_lock);
|
|
for (auto& ptr : _list)
|
|
v.push_back(ptr);
|
|
}
|
|
|
|
for (auto& ptr : v)
|
|
func(ptr);
|
|
}
|
|
|
|
uint32_t Component::Collection::size()
|
|
{
|
|
Poco::FastMutex::ScopedLock lock(_lock);
|
|
return _list.size();
|
|
}
|
|
|
|
Component::Ptr Component::Collection::first()
|
|
{
|
|
return *(_list.begin());
|
|
}
|
|
|
|
/*
|
|
*******************************************************************************
|
|
* CLASS DEFINE
|
|
*******************************************************************************
|
|
*/
|
|
Component::Component(const std::string& name): _name(name), _frameQueues(this), _packetQueues(this), _started(false)
|
|
{
|
|
logger().debug("%s(%#x): CREATE.", name, reinterpret_cast<uintptr_t>(this));
|
|
}
|
|
|
|
Component::~Component()
|
|
{
|
|
logger().debug("%s(%#x): RELEASE.", name(), reinterpret_cast<uintptr_t>(this));
|
|
}
|
|
|
|
Component::Ptr Component::link(Ptr pNext)
|
|
{
|
|
if (pNext.isNull())
|
|
return nullptr;
|
|
|
|
if (nextComponents().add(pNext))
|
|
{
|
|
logger().debug("%s(%#x): ===> %s(%#x)", name(), reinterpret_cast<uintptr_t>(this), pNext->name(),
|
|
reinterpret_cast<uintptr_t>(pNext.get()));
|
|
pNext->doLink(sharedFromThis());
|
|
}
|
|
|
|
return pNext;
|
|
}
|
|
|
|
void Component::unlink(Component::Ptr pNext)
|
|
{
|
|
if (pNext.isNull())
|
|
return;
|
|
|
|
if (nextComponents().remove(pNext))
|
|
{
|
|
logger().debug("%s(%#x): =X=> %s(%#x)", name(), reinterpret_cast<uintptr_t>(this), pNext->name(),
|
|
reinterpret_cast<uintptr_t>(pNext.get()));
|
|
pNext->doUnlink(sharedFromThis());
|
|
}
|
|
}
|
|
|
|
void Component::unlink(bool unlinkAllPrevious, bool unlinkAllNext)
|
|
{
|
|
if (unlinkAllPrevious)
|
|
{
|
|
previousComponents().foreach ([this](Ptr pObject) -> void { pObject->unlink(sharedFromThis()); });
|
|
}
|
|
|
|
if (unlinkAllNext)
|
|
{
|
|
nextComponents().foreach ([this](Ptr pObject) -> void { unlink(pObject); });
|
|
}
|
|
}
|
|
|
|
void Component::enable(const void* pUser, const std::string& desc)
|
|
{
|
|
logger().debug("%s(%#x): Enabled by %s(%#x)", name(), reinterpret_cast<uintptr_t>(this), desc,
|
|
reinterpret_cast<uintptr_t>(pUser));
|
|
Poco::Mutex::ScopedLock lock(_componentsLock);
|
|
bool shouldStart = (userCount() == 0);
|
|
incrementUserRefCount(pUser);
|
|
if (shouldStart)
|
|
{
|
|
frameQueues().flush();
|
|
packetQueues().flush();
|
|
|
|
initialize();
|
|
|
|
previousComponents().foreach ([&](Ptr ptr) -> void { ptr->enable(this); });
|
|
|
|
logger().debug("%s(%#x): START.", name(), reinterpret_cast<uintptr_t>(this));
|
|
start();
|
|
_traffic = 0;
|
|
_started = true;
|
|
}
|
|
}
|
|
|
|
void Component::disable(const void* pUser, const std::string& desc, bool ignoreRefCount)
|
|
{
|
|
logger().debug("%s(%#x): Disabled by %s(%#x)", name(), reinterpret_cast<uintptr_t>(this), desc,
|
|
reinterpret_cast<uintptr_t>(pUser));
|
|
Poco::Mutex::ScopedLock lock(_componentsLock);
|
|
if (!started())
|
|
{
|
|
logger().debug("%s(%#x): Stop already.", name(), reinterpret_cast<uintptr_t>(this));
|
|
return;
|
|
}
|
|
|
|
if (ignoreRefCount)
|
|
eraseUser(pUser);
|
|
else
|
|
decrementUserRefCount(pUser);
|
|
|
|
if (userCount() == 0)
|
|
{
|
|
logger().debug("%s(%#x): STOP.", name(), reinterpret_cast<uintptr_t>(this));
|
|
stop();
|
|
deinitialize();
|
|
_traffic = 0;
|
|
_started = false;
|
|
|
|
previousComponents().foreach ([&](Component::Ptr ptr) -> void { ptr->disable(this); });
|
|
|
|
frameQueues().flush();
|
|
packetQueues().flush();
|
|
}
|
|
}
|
|
|
|
int Component::userCount()
|
|
{
|
|
Poco::Mutex::ScopedLock lock(_componentsLock);
|
|
return _users.size();
|
|
}
|
|
|
|
int Component::nextComponentCount()
|
|
{
|
|
return nextComponents().size();
|
|
}
|
|
|
|
int Component::previousComponentCount()
|
|
{
|
|
return previousComponents().size();
|
|
}
|
|
|
|
QueueStatus Component::queueStatus(Ptr who)
|
|
{
|
|
{
|
|
auto pQueue = packetQueues().find(who);
|
|
if (pQueue)
|
|
return pQueue->status();
|
|
}
|
|
|
|
{
|
|
auto pQueue = frameQueues().find(who);
|
|
if (pQueue)
|
|
return pQueue->status();
|
|
}
|
|
|
|
return QueueStatus::UNKNOWN;
|
|
}
|
|
|
|
void Component::initialize()
|
|
{
|
|
}
|
|
|
|
void Component::deinitialize()
|
|
{
|
|
}
|
|
|
|
void Component::doLink(Ptr pPrevious)
|
|
{
|
|
if (previousComponents().add(pPrevious))
|
|
{
|
|
onLink(pPrevious);
|
|
if (started())
|
|
pPrevious->enable(this);
|
|
}
|
|
}
|
|
|
|
void Component::onLink(Ptr pPrevious)
|
|
{
|
|
}
|
|
|
|
void Component::doUnlink(Ptr pPrevious)
|
|
{
|
|
if (previousComponents().remove(pPrevious))
|
|
{
|
|
onUnlink(pPrevious);
|
|
pPrevious->disable(this, true);
|
|
}
|
|
}
|
|
|
|
void Component::onUnlink(Ptr pPrevious)
|
|
{
|
|
}
|
|
|
|
void Component::post(Ptr pSender, Frame::Ptr pFrame, MediaType type, bool force)
|
|
{
|
|
post(pSender, frameQueues(), pFrame, type, force);
|
|
}
|
|
|
|
void Component::post(Ptr pSender, Packet::Ptr pPacket, MediaType type, bool force)
|
|
{
|
|
post(pSender, packetQueues(), pPacket->clone(), type, force);
|
|
}
|
|
|
|
void Component::notify(Notification::Ptr pNotification, bool shouldPropagate)
|
|
{
|
|
_notificationCenter.postNotification(pNotification);
|
|
if (shouldPropagate)
|
|
{
|
|
logger().debug("%s(Sender is '%s'): Notify '%s' to next components", name(), pNotification->getSender()->name(),
|
|
pNotification->name());
|
|
nextComponents().foreach (
|
|
[&](Component::Ptr ptr) -> void
|
|
{
|
|
if (ptr->started())
|
|
ptr->notify(pNotification, shouldPropagate);
|
|
});
|
|
}
|
|
}
|
|
|
|
void Component::notifyNextComponents(Ptr pSender, Notification::Ptr pNotification, bool shouldPropagate)
|
|
{
|
|
logger().debug("%s: Notify '%s' to next components", pSender->name(), pNotification->name());
|
|
pNotification->setSender(pSender);
|
|
nextComponents().foreach (
|
|
[&](Component::Ptr ptr) -> void
|
|
{
|
|
if (ptr->started())
|
|
ptr->notify(pNotification, shouldPropagate);
|
|
});
|
|
}
|
|
|
|
void Component::incrementUserRefCount(const void* pUser)
|
|
{
|
|
if (_users.find(pUser) == _users.end())
|
|
_users[pUser] = 0;
|
|
|
|
_users[pUser]++;
|
|
}
|
|
|
|
void Component::decrementUserRefCount(const void* pUser)
|
|
{
|
|
if (_users.find(pUser) != _users.end())
|
|
{
|
|
_users[pUser]--;
|
|
if (_users[pUser] <= 0)
|
|
_users.erase(pUser);
|
|
}
|
|
}
|
|
|
|
void Component::eraseUser(const void* pUser)
|
|
{
|
|
_users.erase(pUser);
|
|
}
|
|
|
|
bool Component::syncWait(uint64_t ts, std::atomic<bool>& isCancelled)
|
|
{
|
|
uint64_t now = System::timestamp();
|
|
while (ts > now && !isCancelled)
|
|
{
|
|
now = System::timestamp();
|
|
usleep(100);
|
|
}
|
|
|
|
return (now >= ts);
|
|
}
|
|
|
|
bool Component::syncWait(uint64_t ts, uint32_t maxWaitTime)
|
|
{
|
|
const uint64_t startTime = System::timestamp();
|
|
const uint64_t timeout = maxWaitTime * 1000;
|
|
while (true)
|
|
{
|
|
uint64_t now = System::timestamp();
|
|
|
|
if (now >= ts)
|
|
return true;
|
|
|
|
if (now - startTime >= timeout)
|
|
return false;
|
|
|
|
const uint64_t remaining = ::Util::Math::min(ts - now, timeout - (now - startTime));
|
|
const useconds_t us = ::Util::Math::clamp(remaining / 10, 1000ULL, 100000ULL);
|
|
usleep(us);
|
|
}
|
|
}
|
|
|
|
void Component::publishEvent(Event event, bool sync)
|
|
{
|
|
if (sync)
|
|
EventOccurred.notify(this, event);
|
|
else
|
|
EventOccurred.notifyAsync(this, event);
|
|
}
|
|
}
|
|
}
|