p26001-store-acs-soc-fw-app/video/core/Demuxer.cpp

/*
 *******************************************************************************
 *                                   INCLUDE
 *******************************************************************************
 */
#include "Demuxer.h"
#include "System.h"
#include "Logger.h"
#include "util/Broadcast.h"
#include <Poco/Format.h>
#include <unistd.h>

namespace Video
{
namespace Core
{
/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
class DummyVideoEncoder: public VideoEncoder
{
public:
    DummyVideoEncoder(const std::string& name): VideoEncoder(name, CodecID::H264)
    {
    }

protected:
    void onConfigChanged(ParameterNames& names) override
    {
        if (names.exists(_CodecID))
            serialCodeManager().put(_CodecID, getCodecID());

        if (names.exists(_Width))
            serialCodeManager().put(_Width, getWidth());

        if (names.exists(_Height))
            serialCodeManager().put(_Height, getHeight());

        if (names.exists(_FrameRate))
            serialCodeManager().put(_FrameRate, getFrameRate());
    }
};

class DummyAudioEncoder: public AudioEncoder
{
public:
    DummyAudioEncoder(const std::string& name): AudioEncoder(name, CodecID::AAC)
    {
    }

protected:
    Core::Packet::Ptr encode() override
    {
        return nullptr;
    }

    void onConfigChanged(ParameterNames& names) override
    {
        if (names.exists(_CodecID))
            serialCodeManager().put(_CodecID, getCodecID());

        if (names.exists(_SampleRate))
            serialCodeManager().put(_SampleRate, getSampleRate());

        if (names.exists(_SampleFormat))
            serialCodeManager().put(_SampleFormat, getSampleFormat());
    }
};

/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
DemuxHandler::DemuxHandler(const std::string& name, const std::string& uri)
    : _name(name), _uri(uri), _scheme(Poco::URI(uri).getScheme()), _timeDrift(0)
{
}

DemuxHandler::~DemuxHandler()
{
}

int64_t DemuxHandler::getTimeDrift(uint64_t ts)
{
    if (0 == _timeDrift)
        _timeDrift = getBaseline() - ts;

    return _timeDrift;
}

void DemuxHandler::resetTimeDrift()
{
    _timeDrift = 0;
}

/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
class DemuxerHandlerController: public Util::BroadcastReceiver, public Util::BroadcastReceiverRegistry
{
public:
    DemuxerHandlerController(DemuxHandler::Ptr pHandler, int index, std::atomic<bool>& isCancelled)
        : _skip(false), _newIndex(0), _newPosition(0), _pHandler(pHandler), _index(index), _isCancelled(isCancelled)
    {
        Util::IntentFilter* pFilter = new Util::IntentFilter;
        pFilter->addAction("carousel.position");
        registerReceiver(*this, pFilter);
    }

    ~DemuxerHandlerController()
    {
        unregisterReceiver(*this);
    }

protected:
    void onReceive(Util::Intent::Ptr pIntent) override
    {
        int index    = pIntent->getExtra<int>("index");
        int position = pIntent->getExtra<int>("position");
        if (index < 0 || position < 0)
            return;

        uint64_t ts = position * 1000000u;

        if (index != _index)
        {
            _skip        = true;
            _newIndex    = index;
            _newPosition = ts;

            _isCancelled = true;

            return;
        }

        _pHandler->seek(ts);
    }

    READONLY_PROPERTY_DECLARE(skip, bool)
    READONLY_PROPERTY_DECLARE(newIndex, int)
    READONLY_PROPERTY_DECLARE(newPosition, uint64_t)

private:
    DemuxHandler::Ptr  _pHandler;
    int                _index;
    std::atomic<bool>& _isCancelled;
};

/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
Demuxer::Demuxer(const std::string& name)
    : Component(name), _pVideoEncoder(new DummyVideoEncoder(name + "/VideoEncoder")),
      _pAudioEncoder(new DummyAudioEncoder(name + "/AudioEncoder"))
{
    setUsingTCPTransport(false);
    setURICollection({});
}

Demuxer::~Demuxer()
{
}

#define __try_run(x) \
    try \
    { \
        x \
    } \
    catch (Poco::Exception & e) \
    { \
        logger().error("%s: %s", name(), e.displayText()); \
        Poco::Thread::sleep(2000); \
    } \
    catch (std::exception & e) \
    { \
        logger().error("%s: %s", name(), std::string(e.what())); \
        Poco::Thread::sleep(2000); \
    }

void Demuxer::start()
{
    if (getURICollection().size() == 0)
    {
        logger().warning("%s: URI collection is empty!", name());
        return;
    }

    _task.run(
        [this](std::atomic<bool>& isCancelled) -> void
        {
            _playIndex = 0;
            PlayIndexChanged.notify(this);

            int index = 0;

            uint64_t position = 0;

            uint64_t timeline = 1000000;
            while (!isCancelled)
            {
                std::string       uri;
                DemuxHandler::Ptr pHandler;

                __try_run({ uri = getURICollection().at(index); });
                __try_run({ pHandler = DemuxHandler::create(uri); });

                if (!pHandler.isNull())
                {
                    __try_run({
                        _playPosition = 0;

                        if (Poco::URI(uri).getScheme() == "rtsp")
                            pHandler->setUsingTCPTransport(getUsingTCPTransport());
                        pHandler->setBaseline(timeline);

                        Poco::SharedPtr<DemuxerHandlerController> pController;
                        if (Poco::URI(uri).getScheme() == "file")
                        {
                            pController = new DemuxerHandlerController(pHandler, index, isCancelled);
                            if (position)
                            {
                                pHandler->seek(position);
                                position = 0;
                            }
                        }

                        pHandler->run(
                            videoEncoder(), audioEncoder(),
                            [&](Packet::Ptr pPacket, MediaType type) -> void
                            {
                                if (postToNextComponents(sharedFromThis(), pPacket, type, isCancelled))
                                {
                                    int64_t pos  = pHandler->getPosition();
                                    int64_t diff = std::abs(pos - _playPosition);
                                    if (diff >= 1000000)
                                    {
                                        _playPosition = pos;
                                        PlayPositionChanged.notify(this);
                                    }
                                }
                            },
                            isCancelled);

                        timeline = pHandler->getTimeline();

                        if (pController && pController->skip())
                        {
                            isCancelled = false;

                            index    = pController->newIndex();
                            position = pController->newPosition();

                            _playIndex    = index;
                            _playPosition = position;

                            PlayIndexChanged.notify(this);
                            PlayPositionChanged.notify(this);

                            continue;
                        }
                    });
                }
                else
                    logger().error("%s: Not implemented URI scheme '%s'", name(), Poco::URI(uri).getScheme());

                if (++index >= (int)getURICollection().size())
                    index = 0;

                if (_playIndex != index)
                {
                    _playIndex = index;
                    PlayIndexChanged.notify(this);
                }
            }
        });
}

void Demuxer::stop()
{
    _task.cancel();
}

void Demuxer::onLink(Component::Ptr pPrevious)
{
    throw Poco::LogicException("Not supported for link");
}

void Demuxer::onUnlink(Component::Ptr pPrevious)
{
    throw Poco::LogicException("Not supported for link");
}

void Demuxer::onConfigChanged(ParameterNames& names)
{
    if (!started())
        return;

    if (names.exists(_UsingTCPTransport) || names.exists(_URICollection))
    {
        stop();
        start();
    }
}
}
}