/*
 *******************************************************************************
 *                                   INCLUDE
 *******************************************************************************
 */
#include "StreamPusher.h"
#include "Service.h"
#include "Logger.h"
#include <Poco/Exception.h>

namespace Video
{
namespace ZLMediaKit
{
/*
 *******************************************************************************
 *                                CLASS DEFINE
 *******************************************************************************
 */
StreamPusher::StreamPusher(const std::string& name, const std::string& streamApp, const std::string streamID)
    : _name(name + "/ZLStreamPusher"), _streamApp(streamApp), _streamID(streamID), _task(name)
{
    setConnected(false);

    setURL("");
    setUsingTCPTransport(false);
    setRetryInterval(2000);
}

StreamPusher::~StreamPusher()
{
}

void StreamPusher::start()
{
    if (streamApp().empty() || streamID().empty() || getURL().empty())
        throw Poco::InvalidArgumentException(_name + ": stream-app, stream-id or URL empty");

    _task.run(
        [this](std::atomic<bool>& isCancelled) -> void
        {
            setStarted(true);
            const std::string scheme = Poco::URI(getURL()).getScheme();
            while (!isCancelled)
            {
                setConnected(false);

                while (!isCancelled)
                {
                    try
                    {
                        auto result = Service::addStreamPusherProxy(scheme, streamApp(), streamID(), getURL());
                        int  code   = result.get<0>();
                        if (code == 0)
                        {
                            setConnected(true);

                            _key = result.get<1>();
                            ConnectionChanged.notify(this);
                            break;
                        }
                    }
                    catch (Poco::Exception& e)
                    {
                        logger().error("%s: Create %s stream pusher FAIL: %s", name(), scheme, e.displayText());
                    }
                    catch (std::exception& e)
                    {
                        logger().error("%s: Create %s stream pusher FAIL: %s", name(), scheme, (std::string)e.what());
                    }

                    Poco::Thread::sleep(getRetryInterval());
                }

                while (!isCancelled)
                {
                    Poco::Thread::sleep(2000);
                    Service::StremPusherProxyInfo info;
                    info.status = 1;
                    Service::getStremPusherProxyInfo(_key, info);
                    bool connected = (info.status == 0);
                    if (connected != getConnected())
                    {
                        setConnected(connected);
                        ConnectionChanged.notify(this);
                    }
                    if (info.code != 0 && info.msg == "can not find pusher")
                    {
                        break;
                    }
                }
            }

            Service::delStreamPusherProxy(_key);
            if (getConnected())
            {
                setConnected(false);
                ConnectionChanged.notify(this);
            }

            setStarted(false);
        });
}

void StreamPusher::cancel()
{
    _task.cancel();
}

void StreamPusher::onConfigChanged(ParameterNames& names)
{
    if (!getStarted())
        return;

    if (names.exists(_URL) || names.exists(_UsingTCPTransport))
    {
        cancel();
        start();
    }
}
}
}