#ifndef AB01F7C9_DADD_4ADA_88F6_ED371CF36813 #define AB01F7C9_DADD_4ADA_88F6_ED371CF36813 /* ******************************************************************************* * INCLUDE ******************************************************************************* */ #include "util/Configurable.h" #include "util/ClassAssistant.h" #include "Frame.h" #include "Packet.h" #include #include #include #include #include #include #include #include #include #include #include #include namespace Video { namespace Core { /* ******************************************************************************* * CLASS DECLARE ******************************************************************************* */ class Component: public Poco::RefCountedObject, public Util::Configurable { public: enum class Event { IMAGE_SIZE_CHANGED, }; using Ptr = Poco::AutoPtr; class Collection { public: bool add(Ptr); bool remove(Ptr); void foreach (std::function func); uint32_t size(); Ptr first(); private: std::list _list; Poco::FastMutex _lock; }; class Notification: public Poco::Notification { public: using Ptr = Poco::AutoPtr; public: PROPERTY_DECLARE(Sender, Component::Ptr, nullptr, public, public); std::string name() const override { return Util::ClassAssistant::demangle(*this); } }; template class Queues { public: Queues(Ptr pOwner); void add(Ptr, const std::string& name, MediaType, int limit = 300); void remove(Ptr); void clear(); bool exists(Ptr); bool empty(); typename QT::Ptr first(); typename QT::Ptr find(Ptr); Ptr firstComponent(); void flush(); template void foreach (const F&& func); private: Ptr _pOwner; Poco::FastMutex _mutex; std::map _queues; }; public: Component(const std::string& name); Component(const Component&) = delete; virtual ~Component(); public: const std::string& name() const; Ptr link(Ptr pNext); void unlink(Ptr pNext); void unlink(bool unlinkAllPrevious, bool unlinkAllNext); template void enable(T* pObject); template void disable(T* pObject, bool ignoreRefCount = false); template bool enabled(T* pObject); Ptr sharedFromThis(); void enable(const void* pUser, const std::string& desc); void disable(const void* pUser, const std::string& desc, bool ignoreRefCount = false); int userCount(); int nextComponentCount(); int previousComponentCount(); const SerialCode& code(); QueueStatus queueStatus(Ptr who); virtual uint32_t traffic(); bool started() const; Poco::BasicEvent EventOccurred; protected: virtual void start() = 0; virtual void stop() = 0; virtual void initialize(); virtual void deinitialize(); virtual void doLink(Ptr pPrevious); virtual void onLink(Ptr pPrevious); virtual void doUnlink(Ptr pPrevious); virtual void onUnlink(Ptr pPrevious); Collection& previousComponents(); Collection& nextComponents(); SerialCodeManager& serialCodeManager(); SerialCodeWatcher& serialCodeWatcher(); void post(Ptr pSender, Frame::Ptr pFrame, MediaType type, bool force); void post(Ptr pSender, Packet::Ptr pPacket, MediaType type, bool force); template void post(Ptr pSender, TQueue& queues, TObjectPtr pObject, MediaType type, bool force); template void postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type); template bool postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type, std::atomic& isCancelled); template void registerNotificationHandle(C& object, const H& handle); template void unregisterNotificationHandle(C& object, const H& handle); void notify(Notification::Ptr pNotification, bool shouldPropagate); void notifyNextComponents(Ptr pSender, Notification::Ptr pNotification, bool shouldPropagate = false); Queues& frameQueues(); Queues& packetQueues(); void incrementUserRefCount(const void* pUser); void decrementUserRefCount(const void* pUser); void eraseUser(const void* pUser); bool syncWait(uint64_t ts, std::atomic& isCancelled); bool syncWait(uint64_t ts, uint32_t maxWaitTime); void publishEvent(Event event, bool sync = true); protected: uint32_t _traffic; private: const std::string _name; Collection _previousComponents; Collection _nextComponents; Poco::Mutex _componentsLock; Queues _frameQueues; Queues _packetQueues; std::unordered_map _users; bool _started; SerialCodeManager _serialCodeManager; SerialCodeWatcher _serialCodeWatcher; Poco::NotificationCenter _notificationCenter; }; template inline Component::Queues::Queues(Ptr pOwner): _pOwner(pOwner) { } template void Component::Queues::add(Ptr pComponent, const std::string& name, MediaType type, int limit) { if (exists(pComponent)) throw Poco::ExistsException(_pOwner->name() + "'s queue: " + pComponent->name() + " EXISTS!"); Poco::FastMutex::ScopedLock lock(_mutex); _queues[pComponent] = new QT(name, type, limit); } template void Component::Queues::remove(Ptr pComponent) { Poco::FastMutex::ScopedLock lock(_mutex); _queues.erase(pComponent); } template inline void Component::Queues::clear() { Poco::FastMutex::ScopedLock lock(_mutex); return _queues.clear(); } template inline bool Component::Queues::exists(Ptr pComponent) { Poco::FastMutex::ScopedLock lock(_mutex); return _queues.find(pComponent) != _queues.end(); } template inline bool Component::Queues::empty() { Poco::FastMutex::ScopedLock lock(_mutex); return _queues.empty(); } template inline typename QT::Ptr Component::Queues::first() { Poco::FastMutex::ScopedLock lock(_mutex); if (_queues.empty()) return nullptr; return _queues.begin()->second; } template inline typename QT::Ptr Component::Queues::find(Ptr pComponent) { Poco::FastMutex::ScopedLock lock(_mutex); const auto it = _queues.find(pComponent); if (it != _queues.end()) return it->second; return nullptr; } template inline Component::Ptr Component::Queues::firstComponent() { Poco::FastMutex::ScopedLock lock(_mutex); if (_queues.empty()) return nullptr; return _queues.begin()->first; } template inline void Component::Queues::flush() { Poco::FastMutex::ScopedLock lock(_mutex); for (auto& q : _queues) q.second->clear(); } template template inline void Component::Queues::foreach (const F&& func) { Poco::FastMutex::ScopedLock lock(_mutex); for (auto& q : _queues) func(q.first, q.second); } inline const std::string& Component::name() const { return _name; } inline bool Component::started() const { return _started; } inline const SerialCode& Component::code() { return serialCodeManager().code(); } inline uint32_t Component::traffic() { return _traffic; } inline Component::Collection& Component::previousComponents() { return _previousComponents; } inline Component::Collection& Component::nextComponents() { return _nextComponents; } inline SerialCodeManager& Component::serialCodeManager() { return _serialCodeManager; } inline SerialCodeWatcher& Component::serialCodeWatcher() { return _serialCodeWatcher; } inline Component::Queues& Component::frameQueues() { return _frameQueues; } inline Component::Queues& Component::packetQueues() { return _packetQueues; } template inline void Component::enable(T* pObject) { enable(pObject, std::is_base_of::value ? reinterpret_cast(pObject)->name() : Util::ClassAssistant::demangle()); } template inline void Component::disable(T* pObject, bool ignoreRefCount) { disable(pObject, std::is_base_of::value ? reinterpret_cast(pObject)->name() : Util::ClassAssistant::demangle(), ignoreRefCount); } template inline bool Video::Core::Component::enabled(T* pUser) { Poco::Mutex::ScopedLock lock(_componentsLock); return (_users.find(pUser) != _users.end()); } template inline void Component::post(Ptr pSender, TQueue& queues, TObjectPtr pObject, MediaType type, bool force) { auto pQueue = queues.find(pSender); if (pQueue && pQueue->type() == type) { pQueue->enqueue(pObject, force); _traffic++; } } template inline void Component::postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type) { nextComponents().foreach ( [&](Ptr ptr) -> void { if (ptr->started()) ptr->post(pSender, pObject, type, true); }); } template inline bool Component::postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type, std::atomic& isCancelled) { int count = 0; int num = 0; nextComponents().foreach ( [&](Ptr ptr) -> void { if (ptr->started()) { num++; do { try { ptr->post(pSender, pObject, type, false); count++; break; } catch (...) { if (!isCancelled) ::usleep(1000); } } while (!isCancelled); } }); return count == num; } template inline void Component::registerNotificationHandle(C& object, const H& handle) { _notificationCenter.addObserver(Poco::NObserver(object, handle)); } template inline void Component::unregisterNotificationHandle(C& object, const H& handle) { _notificationCenter.removeObserver(Poco::NObserver(object, handle)); } inline Component::Ptr Component::sharedFromThis() { return Ptr(this, true); } } } #endif