122 lines
3.6 KiB
C++
122 lines
3.6 KiB
C++
/*
|
|
*******************************************************************************
|
|
* 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();
|
|
}
|
|
}
|
|
}
|
|
}
|