p26001-store-acs-soc-fw-app/video/core/Component.h

#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 <Poco/AutoPtr.h>
#include <Poco/Mutex.h>
#include <Poco/NotificationCenter.h>
#include <Poco/NObserver.h>
#include <Poco/RefCountedObject.h>
#include <Poco/SharedPtr.h>
#include <Poco/BasicEvent.h>
#include <list>
#include <unordered_map>
#include <functional>
#include <type_traits>
#include <unistd.h>

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<Component>;
    class Collection
    {
    public:
        bool add(Ptr);
        bool remove(Ptr);
        void foreach (std::function<void(Ptr)> func);
        uint32_t size();
        Ptr      first();

    private:
        std::list<Ptr>  _list;
        Poco::FastMutex _lock;
    };

    class Notification: public Poco::Notification
    {
    public:
        using Ptr = Poco::AutoPtr<Notification>;

    public:
        PROPERTY_DECLARE(Sender, Component::Ptr, nullptr, public, public);

        std::string name() const override
        {
            return Util::ClassAssistant::demangle(*this);
        }
    };

    template <class QT> 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 <typename F> void foreach (const F&& func);

    private:
        Ptr                             _pOwner;
        Poco::FastMutex                 _mutex;
        std::map<Ptr, typename QT::Ptr> _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 <class T> void enable(T* pObject);
    template <class T> void disable(T* pObject, bool ignoreRefCount = false);
    template <class T> 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<Event> 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 <class TQueue, class TObjectPtr>
    void post(Ptr pSender, TQueue& queues, TObjectPtr pObject, MediaType type, bool force);

    template <class TObjectPtr> void postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type);

    template <class TObjectPtr>
    bool postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type, std::atomic<bool>& isCancelled);

    template <class N, class C, class H> void registerNotificationHandle(C& object, const H& handle);
    template <class N, class C, class H> 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<FrameQueue>&  frameQueues();
    Queues<PacketQueue>& packetQueues();

    void incrementUserRefCount(const void* pUser);
    void decrementUserRefCount(const void* pUser);
    void eraseUser(const void* pUser);

    bool syncWait(uint64_t ts, std::atomic<bool>& 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<FrameQueue>                   _frameQueues;
    Queues<PacketQueue>                  _packetQueues;
    std::unordered_map<const void*, int> _users;
    bool                                 _started;
    SerialCodeManager                    _serialCodeManager;
    SerialCodeWatcher                    _serialCodeWatcher;
    Poco::NotificationCenter             _notificationCenter;
};

template <class QT> inline Component::Queues<QT>::Queues(Ptr pOwner): _pOwner(pOwner)
{
}

template <class QT> void Component::Queues<QT>::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 <class QT> void Component::Queues<QT>::remove(Ptr pComponent)
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    _queues.erase(pComponent);
}

template <class QT> inline void Component::Queues<QT>::clear()
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    return _queues.clear();
}

template <class QT> inline bool Component::Queues<QT>::exists(Ptr pComponent)
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    return _queues.find(pComponent) != _queues.end();
}

template <class QT> inline bool Component::Queues<QT>::empty()
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    return _queues.empty();
}

template <class QT> inline typename QT::Ptr Component::Queues<QT>::first()
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    if (_queues.empty())
        return nullptr;

    return _queues.begin()->second;
}

template <class QT> inline typename QT::Ptr Component::Queues<QT>::find(Ptr pComponent)
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    const auto                  it = _queues.find(pComponent);
    if (it != _queues.end())
        return it->second;

    return nullptr;
}

template <class QT> inline Component::Ptr Component::Queues<QT>::firstComponent()
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    if (_queues.empty())
        return nullptr;

    return _queues.begin()->first;
}

template <class QT> inline void Component::Queues<QT>::flush()
{
    Poco::FastMutex::ScopedLock lock(_mutex);
    for (auto& q : _queues)
        q.second->clear();
}

template <class QT> template <typename F> inline void Component::Queues<QT>::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<FrameQueue>& Component::frameQueues()
{
    return _frameQueues;
}

inline Component::Queues<PacketQueue>& Component::packetQueues()
{
    return _packetQueues;
}

template <class T> inline void Component::enable(T* pObject)
{
    enable(pObject, std::is_base_of<Component, T>::value ? reinterpret_cast<Component*>(pObject)->name()
                                                         : Util::ClassAssistant::demangle<T>());
}

template <class T> inline void Component::disable(T* pObject, bool ignoreRefCount)
{
    disable(pObject,
            std::is_base_of<Component, T>::value ? reinterpret_cast<Component*>(pObject)->name()
                                                 : Util::ClassAssistant::demangle<T>(),
            ignoreRefCount);
}

template <class T> inline bool Video::Core::Component::enabled(T* pUser)
{
    Poco::Mutex::ScopedLock lock(_componentsLock);
    return (_users.find(pUser) != _users.end());
}

template <class TQueue, class TObjectPtr>
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 <class TObjectPtr> 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 <class TObjectPtr>
inline bool Component::postToNextComponents(Ptr pSender, TObjectPtr pObject, MediaType type,
                                            std::atomic<bool>& 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 <class N, class C, class H> inline void Component::registerNotificationHandle(C& object, const H& handle)
{
    _notificationCenter.addObserver(Poco::NObserver<C, N>(object, handle));
}

template <class N, class C, class H> inline void Component::unregisterNotificationHandle(C& object, const H& handle)
{
    _notificationCenter.removeObserver(Poco::NObserver<C, N>(object, handle));
}

inline Component::Ptr Component::sharedFromThis()
{
    return Ptr(this, true);
}
}
}
#endif