Skip to content

E
evsuits
提交 6dd4d49b authored 作者: blu's avatar blu
浏览文件
操作

bugfix: check 0 sized file

上级 ec900f1c
隐藏空白字符变更
内嵌 并排
正在显示 包含 176 行增加 和 112 行删除
opencv-motion-detect/evcloudsvc.cpp
浏览文件 @ 6dd4d49b
...@@ -100,10 +100,11 @@ private: ...@@ -100,10 +100,11 @@ private:
ret = z_send_multiple(pRouter, v); ret = z_send_multiple(pRouter, v);
if(ret <0) { if(ret <0) {
spdlog::error("evcloudsvc failed to send config to {}", sn); spdlog::error("evcloudsvc failed to send config to {}", sn);
}else{ }
else {
spdlog::info("evcloudsvc config sent to {}: {}", sn, cfg); spdlog::info("evcloudsvc config sent to {}: {}", sn, cfg);
} }
//} //}
return ret; return ret;
...@@ -373,7 +374,7 @@ private: ...@@ -373,7 +374,7 @@ private:
if(data["code"] != 0) { if(data["code"] != 0) {
json resp; json resp;
resp["target"] = selfId, resp["target"] = selfId,
resp["metaType"] = EV_MSG_META_PONG; resp["metaType"] = EV_MSG_META_PONG;
resp["data"] = data["msg"]; resp["data"] = data["msg"];
sendEdgeMsg(resp); sendEdgeMsg(resp);
} }
...@@ -482,7 +483,8 @@ private: ...@@ -482,7 +483,8 @@ private:
spdlog::error("evcloudsvc {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno())); spdlog::error("evcloudsvc {} failed to send multiple: {}", devSn, zmq_strerror(zmq_errno()));
} }
} }
}else{ }
else {
json resp; json resp;
resp["metaType"] = EV_MSG_META_PONG; resp["metaType"] = EV_MSG_META_PONG;
resp["target"] = selfId; resp["target"] = selfId;
...@@ -561,12 +563,13 @@ private: ...@@ -561,12 +563,13 @@ private:
// eventToSlicer["end"] // eventToSlicer["end"]
// eventToSlicer["sender"] = selfId; // eventToSlicer["sender"] = selfId;
json sendEdgeMsg(json &body) { json sendEdgeMsg(json &body)
{
json ret; json ret;
ret["code"] = 0; ret["code"] = 0;
ret["msg"] = "ok"; ret["msg"] = "ok";
string msg; string msg;
try{ try {
auto target = body["target"].get<string>(); auto target = body["target"].get<string>();
auto v = strutils::split(target, ':'); auto v = strutils::split(target, ':');
if(v.size() == 1 || v.size() == 3) { if(v.size() == 1 || v.size() == 3) {
...@@ -574,26 +577,30 @@ private: ...@@ -574,26 +577,30 @@ private:
meta["type"] = body["metaType"]; meta["type"] = body["metaType"];
if(body.count("metaValue") == 0) { if(body.count("metaValue") == 0) {
// meta["value"] = ""; // meta["value"] = "";
}else{ }
else {
meta["value"] = body["metaValue"]; meta["value"] = body["metaValue"];
} }
body["sender"] = devSn; body["sender"] = devSn;
if(peerData["status"].count(v[0]) == 0 || peerData["status"][v[0]] == 0){ if(peerData["status"].count(v[0]) == 0 || peerData["status"][v[0]] == 0) {
spdlog::warn("evcloudsvc sent msg {} to {}, but it was offline", body.dump(), v[0]); spdlog::warn("evcloudsvc sent msg {} to {}, but it was offline", body.dump(), v[0]);
}else{ }
else {
} }
int i= z_send(pRouter, v[0], devSn, meta, body.dump()); int i= z_send(pRouter, v[0], devSn, meta, body.dump());
if(i < 0) { if(i < 0) {
msg = fmt::format("evcloudsvc failed to z_zend msg: {} :{}",zmq_strerror(zmq_errno()) ,body.dump()); msg = fmt::format("evcloudsvc failed to z_zend msg: {} :{}",zmq_strerror(zmq_errno()),body.dump());
throw StrException(msg); throw StrException(msg);
} }
}else{ }
else {
msg = fmt::format("evcloudsvc invliad target field({}) in body: {}", target, body.dump()); msg = fmt::format("evcloudsvc invliad target field({}) in body: {}", target, body.dump());
throw StrException(msg); throw StrException(msg);
} }
}catch(exception &e) { }
catch(exception &e) {
ret["msg"] = e.what(); ret["msg"] = e.what();
spdlog::error(e.what()); spdlog::error(e.what());
ret["code"] = -1; ret["code"] = -1;
...@@ -602,26 +609,30 @@ private: ...@@ -602,26 +609,30 @@ private:
return ret; return ret;
} }
json handleCmd(json &body){ json handleCmd(json &body)
{
json ret; json ret;
ret["code"] = -1; ret["code"] = -1;
ret["msg"] = "unkown msg"; ret["msg"] = "unkown msg";
spdlog::info("evcloudsvc handle cmd: {}", body.dump()); spdlog::info("evcloudsvc handle cmd: {}", body.dump());
if(body.count("target") != 0 && body["target"].is_string() && body.count("metaType") !=0 && body["metaType"].is_string() && if(body.count("target") != 0 && body["target"].is_string() && body.count("metaType") !=0 && body["metaType"].is_string() &&
body.count("data") != 0 && body["data"].is_object() && body.count("metaValue") !=0 && body["metaValue"].is_string()) { body.count("data") != 0 && body["data"].is_object() && body.count("metaValue") !=0 && body["metaValue"].is_string()) {
// it's msg to edge. // it's msg to edge.
return sendEdgeMsg(body); return sendEdgeMsg(body);
}else{ }
else {
return ret; return ret;
} }
} }
json getReleaseBundle(string bid) { json getReleaseBundle(string bid)
{
json ret; json ret;
return ret; return ret;
} }
json addReleaseBundle(json &bundle) { json addReleaseBundle(json &bundle)
{
json ret; json ret;
return ret; return ret;
} }
...@@ -752,10 +763,11 @@ public: ...@@ -752,10 +763,11 @@ public:
string msg; string msg;
ret["code"] = 0; ret["code"] = 0;
ret["msg"] = "ok"; ret["msg"] = "ok";
try{ try {
auto body = json::parse(req.body); auto body = json::parse(req.body);
ret = this->handleCmd(body); ret = this->handleCmd(body);
}catch(exception &e) { }
catch(exception &e) {
ret["code"] = -1; ret["code"] = -1;
msg = fmt::format("evcloudsvc Post /cmd Exception: {}", e.what()); msg = fmt::format("evcloudsvc Post /cmd Exception: {}", e.what());
spdlog::error(msg); spdlog::error(msg);
...@@ -807,12 +819,13 @@ public: ...@@ -807,12 +819,13 @@ public:
} }
if(this->configMap.count("mod2mgr") ==0 || this->configMap["mod2mgr"].size() ==0) { if(this->configMap.count("mod2mgr") ==0 || this->configMap["mod2mgr"].size() ==0) {
}else{ }
else {
for(auto &k:mods) { for(auto &k:mods) {
this->configMap["mod2mgr"].erase(k); this->configMap["mod2mgr"].erase(k);
} }
} }
this->configMap.erase(sn); this->configMap.erase(sn);
this->peerData.erase(sn); this->peerData.erase(sn);
spdlog::info("evcloudsvc removed sn: {}", sn); spdlog::info("evcloudsvc removed sn: {}", sn);
...@@ -827,11 +840,12 @@ public: ...@@ -827,11 +840,12 @@ public:
string msg; string msg;
ret["code"] = 0; ret["code"] = 0;
ret["msg"] = "ok"; ret["msg"] = "ok";
try{ try {
string bundleId = req.get_param_value("bId"); string bundleId = req.get_param_value("bId");
auto body = json::parse(req.body); auto body = json::parse(req.body);
ret = this->getReleaseBundle(bundleId); ret = this->getReleaseBundle(bundleId);
}catch(exception &e) { }
catch(exception &e) {
ret["code"] = -1; ret["code"] = -1;
msg = fmt::format("evcloudsvc Get /release Exception: {}", e.what()); msg = fmt::format("evcloudsvc Get /release Exception: {}", e.what());
spdlog::error(msg); spdlog::error(msg);
...@@ -846,10 +860,11 @@ public: ...@@ -846,10 +860,11 @@ public:
string msg; string msg;
ret["code"] = 0; ret["code"] = 0;
ret["msg"] = "ok"; ret["msg"] = "ok";
try{ try {
auto body = json::parse(req.body); auto body = json::parse(req.body);
ret = this->addReleaseBundle(body); ret = this->addReleaseBundle(body);
}catch(exception &e) { }
catch(exception &e) {
ret["code"] = -1; ret["code"] = -1;
msg = fmt::format("evcloudsvc Post /release Exception: {}", e.what()); msg = fmt::format("evcloudsvc Post /release Exception: {}", e.what());
spdlog::error(msg); spdlog::error(msg);
...@@ -883,7 +898,7 @@ public: ...@@ -883,7 +898,7 @@ public:
spdlog::error("evcloudsvc failed setup router: {}", addr); spdlog::error("evcloudsvc failed setup router: {}", addr);
exit(1); exit(1);
} }
// setup edge msg processor // setup edge msg processor
thMsgProcessor = thread([this]() { thMsgProcessor = thread([this]() {
while(true) { while(true) {
......
opencv-motion-detect/evdaemon.cpp
浏览文件 @ 6dd4d49b
...@@ -305,7 +305,8 @@ private: ...@@ -305,7 +305,8 @@ private:
// stop // stop
this->peerData["status"][k] = 1; // disabled this->peerData["status"][k] = 1; // disabled
sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, "0"); sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, "0");
}else if(v == 1) { // perm stop }
else if(v == 1) { // perm stop
this->peerData["status"][k] = 2; this->peerData["status"][k] = 2;
this->peerData["config"].erase(k); this->peerData["config"].erase(k);
sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, "0"); sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, "0");
...@@ -330,7 +331,7 @@ private: ...@@ -330,7 +331,7 @@ private:
else { else {
if(int(v) == 3) { if(int(v) == 3) {
sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, to_string(v)); sendCmd2Peer(k, EV_MSG_META_VALUE_CMD_STOP, to_string(v));
} }
} }
} }
else { else {
...@@ -382,18 +383,21 @@ private: ...@@ -382,18 +383,21 @@ private:
spdlog::info("evdaemon {} peer connected: {}", devSn, selfId); spdlog::info("evdaemon {} peer connected: {}", devSn, selfId);
if(this->peerData["tsLastConn"].count(selfId) == 0) { if(this->peerData["tsLastConn"].count(selfId) == 0) {
this->peerData["tsLastConn"][selfId] = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count(); this->peerData["tsLastConn"][selfId] = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count();
}else{ }
else {
if(this->peerData["contConn"].count(selfId) == 0) { if(this->peerData["contConn"].count(selfId) == 0) {
this->peerData["contConn"][selfId] = 0; this->peerData["contConn"][selfId] = 0;
}else{ }
else {
auto delta = this->peerData["contConn"][selfId].get<long>() - chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count(); auto delta = this->peerData["contConn"][selfId].get<long>() - chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count();
if(delta < 3) { // within 3s if(delta < 3) { // within 3s
this->peerData["contConn"][selfId] = this->peerData["contConn"][selfId].get<int>() + 1; this->peerData["contConn"][selfId] = this->peerData["contConn"][selfId].get<int>() + 1;
}else{ }
else {
this->peerData["contConn"][selfId] = 0; this->peerData["contConn"][selfId] = 0;
} }
// refer to startSubsystems // refer to startSubsystems
} }
this->peerData["tsLastConn"][selfId] = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count(); this->peerData["tsLastConn"][selfId] = chrono::duration_cast<chrono::seconds>(chrono::system_clock::now().time_since_epoch()).count();
} }
...@@ -409,9 +413,10 @@ private: ...@@ -409,9 +413,10 @@ private:
spdlog::info("evdaemon {} peer {} config sent: {}", devSn,selfId, cfg); spdlog::info("evdaemon {} peer {} config sent: {}", devSn,selfId, cfg);
} }
else { else {
if(peerData["status"][selfId] == 1 || peerData["status"][selfId] == 2){ if(peerData["status"][selfId] == 1 || peerData["status"][selfId] == 2) {
spdlog::warn("evdaemon {} refuse to start {}: it was asked to be stopped, and is removed from cluster config", this->devSn, selfId); spdlog::warn("evdaemon {} refuse to start {}: it was asked to be stopped, and is removed from cluster config", this->devSn, selfId);
}else{ }
else {
peerData["status"][selfId] = 0; peerData["status"][selfId] = 0;
if(peerData["pids"].count(selfId) != 0) { if(peerData["pids"].count(selfId) != 0) {
peerData["pids"].erase(selfId); peerData["pids"].erase(selfId);
...@@ -420,7 +425,8 @@ private: ...@@ -420,7 +425,8 @@ private:
if(this->bBootstrap) { if(this->bBootstrap) {
spdlog::warn("evdaemon {} peer {} disconnected. restarting it.", devSn, selfId); spdlog::warn("evdaemon {} peer {} disconnected. restarting it.", devSn, selfId);
startSubSystems({selfId}); startSubSystems({selfId});
}else{ }
else {
spdlog::warn("evdaemon {} peer {} disconnected. won't restart it since BOOTSTRAP=false", devSn, selfId); spdlog::warn("evdaemon {} peer {} disconnected. won't restart it since BOOTSTRAP=false", devSn, selfId);
} }
} }
...@@ -538,7 +544,7 @@ private: ...@@ -538,7 +544,7 @@ private:
} }
spdlog::info("evdaemon {} received msg from cloud: {}", devSn, msg); spdlog::info("evdaemon {} received msg from cloud: {}", devSn, msg);
if(v.size() != 3) { if(v.size() != 3) {
spdlog::error("evdaemon {} received invalid msg from evcloudsvc {}", devSn, msg); spdlog::error("evdaemon {} received invalid msg from evcloudsvc {}", devSn, msg);
} }
...@@ -584,46 +590,55 @@ private: ...@@ -584,46 +590,55 @@ private:
spdlog::info("evdaemon {} skip startup subsystems since BOOTSTRAP is set to false", devSn); spdlog::info("evdaemon {} skip startup subsystems since BOOTSTRAP is set to false", devSn);
} }
} }
}else if(meta == EV_MSG_META_TYPE_CMD){ }
else if(meta == EV_MSG_META_TYPE_CMD) {
spdlog::info("evdaemon {} received cmd from cloud: {}", devSn, msg); spdlog::info("evdaemon {} received cmd from cloud: {}", devSn, msg);
if(data.count("target") != 0 && data["target"].is_string() && data.count("metaType") !=0 && data["metaType"].is_string() && if(data.count("target") != 0 && data["target"].is_string() && data.count("metaType") !=0 && data["metaType"].is_string() &&
data.count("data") != 0 && data["data"].is_object() && data.count("metaValue") !=0 && data["metaValue"].is_string()) { data.count("data") != 0 && data["data"].is_object() && data.count("metaValue") !=0 && data["metaValue"].is_string()) {
string target = data["target"]; string target = data["target"];
auto v = strutils::split(target, ':'); auto v = strutils::split(target, ':');
if(v.size() == 1) { if(v.size() == 1) {
spdlog::info("evdaemon {} received msg {} from cloud to itself. TODO: functionality extending points such as debug tunnel", devSn, data.dump()); spdlog::info("evdaemon {} received msg {} from cloud to itself. TODO: functionality extending points such as debug tunnel", devSn, data.dump());
}else if(v.size() == 3){ }
else if(v.size() == 3) {
if(this->peerData["status"].count(target) == 0 || this->peerData["status"][target] == 0 || this->peerData["status"] == -1) { if(this->peerData["status"].count(target) == 0 || this->peerData["status"][target] == 0 || this->peerData["status"] == -1) {
spdlog::error("evdaemon {} received {} msg from cloud to {}: {}, but its offline", devSn, meta, target, data.dump()); spdlog::error("evdaemon {} received {} msg from cloud to {}: {}, but its offline", devSn, meta, target, data.dump());
}else{ }
else {
ret = sendCmd2Peer(target, data["metaValue"], data.dump()); ret = sendCmd2Peer(target, data["metaValue"], data.dump());
if(ret < 0) { if(ret < 0) {
spdlog::error("evdaemon {} failed to send msg to peer {}: {}", devSn, data.dump(), zmq_strerror(zmq_errno())); spdlog::error("evdaemon {} failed to send msg to peer {}: {}", devSn, data.dump(), zmq_strerror(zmq_errno()));
}else{ }
else {
spdlog::info("evdaemon {} successfully relayed {} msg from cloud to {}: {}", devSn, meta, target, data.dump()); spdlog::info("evdaemon {} successfully relayed {} msg from cloud to {}: {}", devSn, meta, target, data.dump());
} }
} }
}else{ }
else {
spdlog::info("well"); spdlog::info("well");
} }
}else{ }
else {
spdlog::info("done"); spdlog::info("done");
} }
}else if(meta == EV_MSG_META_PONG) { }
else if(meta == EV_MSG_META_PONG) {
string info; string info;
if(data.count("data") != 0 ) { if(data.count("data") != 0 ) {
if(data["data"].is_string()) { if(data["data"].is_string()) {
info = fmt::format("evdaemon {} received pong msg from evcloudsvc: {}", devSn, data["data"].get<string>()); info = fmt::format("evdaemon {} received pong msg from evcloudsvc: {}", devSn, data["data"].get<string>());
}else if (data["data"].is_object()) { }
else if (data["data"].is_object()) {
info = fmt::format("evdaemon {} received pong msg from evcloudsvc: {}", devSn, data["data"].dump()); info = fmt::format("evdaemon {} received pong msg from evcloudsvc: {}", devSn, data["data"].dump());
} }
}else{ }
else {
info = fmt::format("evdaemon {} received pong msg from evcloudsvc.", devSn); info = fmt::format("evdaemon {} received pong msg from evcloudsvc.", devSn);
} }
spdlog::info(info); spdlog::info(info);
} }
else{ else {
spdlog::info("evdaemon {} received msg from cloud that having no handler implemented: {}", devSn, msg); spdlog::info("evdaemon {} received msg from cloud that having no handler implemented: {}", devSn, msg);
} }
} }
...@@ -718,7 +733,7 @@ public: ...@@ -718,7 +733,7 @@ public:
{ {
// killall subsystems // killall subsystems
/// peerId -> value /// peerId -> value
peerData["status"] = json(); peerData["status"] = json();
peerData["pids"] = json(); peerData["pids"] = json();
...@@ -822,8 +837,8 @@ public: ...@@ -822,8 +837,8 @@ public:
handleCloudMsg(v); handleCloudMsg(v);
spdlog::info("evdaemon {} successfully handled msg from evcloudsvc", this->devSn); spdlog::info("evdaemon {} successfully handled msg from evcloudsvc", this->devSn);
} }
} }
}); });
......
opencv-motion-detect/evmgr.cpp
浏览文件 @ 6dd4d49b
...@@ -155,7 +155,7 @@ error_exit: ...@@ -155,7 +155,7 @@ error_exit:
this->handleCloudMsg(body); this->handleCloudMsg(body);
} }
}); });
thCloudMsgHandler.detach(); thCloudMsgHandler.detach();
spdlog::info("evmgr {} successfuly inited", devSn); spdlog::info("evmgr {} successfuly inited", devSn);
} }
...@@ -273,7 +273,8 @@ error_exit: ...@@ -273,7 +273,8 @@ error_exit:
} }
bProcessed = true; bProcessed = true;
} catch(exception &e) { }
catch(exception &e) {
bProcessed = false; bProcessed = false;
spdlog::error("evmgr {} exception parse event msg from {} to {}: ", devSn, selfId, peerId, e.what()); spdlog::error("evmgr {} exception parse event msg from {} to {}: ", devSn, selfId, peerId, e.what());
} }
...@@ -300,7 +301,7 @@ error_exit: ...@@ -300,7 +301,7 @@ error_exit:
bProcessed = true; bProcessed = true;
} }
else { else {
try{ try {
json jmeta = json::parse(meta); json jmeta = json::parse(meta);
if(jmeta["type"] == EV_MSG_META_TYPE_BROADCAST) { if(jmeta["type"] == EV_MSG_META_TYPE_BROADCAST) {
if(jmeta.count("value") != 0) { if(jmeta.count("value") != 0) {
...@@ -313,20 +314,23 @@ error_exit: ...@@ -313,20 +314,23 @@ error_exit:
ret = z_send_multiple(pRouter, broadCastMsg); ret = z_send_multiple(pRouter, broadCastMsg);
if(ret < 0) { if(ret < 0) {
spdlog::error("evmgr {} failed to broadcast msg from {} because {}. msg meta: {}", devSn, selfId, zmq_strerror(zmq_errno()), meta); spdlog::error("evmgr {} failed to broadcast msg from {} because {}. msg meta: {}", devSn, selfId, zmq_strerror(zmq_errno()), meta);
}else{ }
else {
spdlog::info("evmgr {} successfully broadcast msg from {} to {}. msg meta: {}", devSn, selfId, k, meta); spdlog::info("evmgr {} successfully broadcast msg from {} to {}. msg meta: {}", devSn, selfId, k, meta);
} }
} }
} }
} }
bProcessed = true; bProcessed = true;
}else if(jmeta["type"] == EV_MSG_META_AVFORMATCTX) { }
else if(jmeta["type"] == EV_MSG_META_AVFORMATCTX) {
bProcessed = true; bProcessed = true;
// ignore; // ignore;
} }
}catch(exception &e) { }
catch(exception &e) {
bProcessed = false; bProcessed = false;
spdlog::error("evmgr {} exception process msg from {} with meta {}: {}", devSn, selfId, meta, e.what()); spdlog::error("evmgr {} exception process msg from {} with meta {}: {}", devSn, selfId, meta, e.what());
} }
} }
} }
...@@ -349,12 +353,12 @@ protected: ...@@ -349,12 +353,12 @@ protected:
bStopSig = true; bStopSig = true;
break; break;
} }
// if(1 == getppid()) { // if(1 == getppid()) {
// spdlog::error("evmgr {} exit since evdaemon is dead", devSn); // spdlog::error("evmgr {} exit since evdaemon is dead", devSn);
// exit(1); // exit(1);
// } // }
auto body = z_recv_multiple(pRouter,false); auto body = z_recv_multiple(pRouter,false);
if(body.size() == 0) { if(body.size() == 0) {
spdlog::error("evmgr {} failed to receive multiple msg: {}", devSn, zmq_strerror(zmq_errno())); spdlog::error("evmgr {} failed to receive multiple msg: {}", devSn, zmq_strerror(zmq_errno()));
......
opencv-motion-detect/evmlmotion.cpp
浏览文件 @ 6dd4d49b
...@@ -159,7 +159,8 @@ private: ...@@ -159,7 +159,8 @@ private:
AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput); AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput);
gotFormat = true; gotFormat = true;
cvMsg.notify_one(); cvMsg.notify_one();
}else{ }
else {
spdlog::warn("evmlmotion {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId); spdlog::warn("evmlmotion {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId);
} }
bProcessed = true; bProcessed = true;
...@@ -223,14 +224,16 @@ private: ...@@ -223,14 +224,16 @@ private:
int portPub = 5556; int portPub = 5556;
if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) { if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) {
portPub = evpuller["portPub"]; portPub = evpuller["portPub"];
}else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()){ }
else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()) {
portPub = evpuller["port-pub"]; portPub = evpuller["port-pub"];
} }
int portRouter = 5550; int portRouter = 5550;
if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) { if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) {
portRouter = evmgr["portRouter"]; portRouter = evmgr["portRouter"];
}else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) { }
else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) {
portRouter = evmgr["port-router"]; portRouter = evmgr["port-router"];
} }
...@@ -646,7 +649,7 @@ protected: ...@@ -646,7 +649,7 @@ protected:
zmq_msg_t msg; zmq_msg_t msg;
AVPacket packet; AVPacket packet;
json eventToSlicer; json eventToSlicer;
// eventToSlicer["type"] = "event"; // eventToSlicer["type"] = "event";
// eventTOSlicer["extraInfo"] = json(); //array // eventTOSlicer["extraInfo"] = json(); //array
// eventToSlicer["start"] // eventToSlicer["start"]
......
opencv-motion-detect/evpuller.cpp
浏览文件 @ 6dd4d49b
...@@ -91,17 +91,19 @@ private: ...@@ -91,17 +91,19 @@ private:
return ret; return ret;
} }
void sendAVInputCtxMsg(string peerId){ void sendAVInputCtxMsg(string peerId)
{
json meta; json meta;
auto msgBody = data2body(const_cast<char*>(pAVFmtCtxBytes), lenAVFmtCtxBytes); auto msgBody = data2body(const_cast<char*>(pAVFmtCtxBytes), lenAVFmtCtxBytes);
if(peerId.empty()){ if(peerId.empty()) {
meta["type"] = EV_MSG_META_TYPE_BROADCAST; meta["type"] = EV_MSG_META_TYPE_BROADCAST;
meta["value"] = EV_MSG_META_AVFORMATCTX; meta["value"] = EV_MSG_META_AVFORMATCTX;
peerId = this->mgrSn + ":evmgr:0"; peerId = this->mgrSn + ":evmgr:0";
}else{ }
else {
meta["type"] = EV_MSG_META_AVFORMATCTX; meta["type"] = EV_MSG_META_AVFORMATCTX;
} }
vector<vector<uint8_t> > rep = {str2body(peerId), str2body(meta.dump()), msgBody}; vector<vector<uint8_t> > rep = {str2body(peerId), str2body(meta.dump()), msgBody};
int ret = z_send_multiple(pDealer, rep); int ret = z_send_multiple(pDealer, rep);
if(ret < 0) { if(ret < 0) {
...@@ -236,14 +238,16 @@ private: ...@@ -236,14 +238,16 @@ private:
int portPub = 5556; int portPub = 5556;
if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) { if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) {
portPub = evpuller["portPub"]; portPub = evpuller["portPub"];
}else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()){ }
else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()) {
portPub = evpuller["port-pub"]; portPub = evpuller["port-pub"];
} }
int portRouter = 5550; int portRouter = 5550;
if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) { if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) {
portRouter = evmgr["portRouter"]; portRouter = evmgr["portRouter"];
}else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) { }
else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) {
portRouter = evmgr["port-router"]; portRouter = evmgr["port-router"];
} }
......
opencv-motion-detect/evpusher.cpp
浏览文件 @ 6dd4d49b
...@@ -94,14 +94,16 @@ private: ...@@ -94,14 +94,16 @@ private:
int portPub = 5556; int portPub = 5556;
if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) { if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) {
portPub = evpuller["portPub"]; portPub = evpuller["portPub"];
}else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()){ }
else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()) {
portPub = evpuller["port-pub"]; portPub = evpuller["port-pub"];
} }
int portRouter = 5550; int portRouter = 5550;
if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) { if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) {
portRouter = evmgr["portRouter"]; portRouter = evmgr["portRouter"];
}else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) { }
else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) {
portRouter = evmgr["port-router"]; portRouter = evmgr["port-router"];
} }
...@@ -260,7 +262,8 @@ private: ...@@ -260,7 +262,8 @@ private:
AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput); AVFormatCtxSerializer::decode((char *)(v[2].data()), v[2].size(), pAVFormatInput);
gotFormat = true; gotFormat = true;
cvMsg.notify_one(); cvMsg.notify_one();
}else{ }
else {
spdlog::warn("evpusher {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId); spdlog::warn("evpusher {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId);
} }
bProcessed = true; bProcessed = true;
......
opencv-motion-detect/evslicer.cpp
浏览文件 @ 6dd4d49b
...@@ -98,9 +98,10 @@ private: ...@@ -98,9 +98,10 @@ private:
gotFormat = true; gotFormat = true;
cvMsg.notify_one(); cvMsg.notify_one();
spdlog::info("evslicer {} got avformat from {}", selfId, peerId); spdlog::info("evslicer {} got avformat from {}", selfId, peerId);
}else{ }
else {
spdlog::warn("evslicer {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId); spdlog::warn("evslicer {} received avformatctx msg from {}, but already proceessed before, ignored. TODO: reinit", selfId, peerId);
} }
} }
else if(meta == EV_MSG_META_EVENT) { else if(meta == EV_MSG_META_EVENT) {
data = json::parse(body2str(v[2])); data = json::parse(body2str(v[2]));
...@@ -168,8 +169,9 @@ private: ...@@ -168,8 +169,9 @@ private:
spdlog::info("evslicer {} received {} cmd from cluster mgr {}", selfId, metaValue, daemonId); spdlog::info("evslicer {} received {} cmd from cluster mgr {}", selfId, metaValue, daemonId);
bProcessed = true; bProcessed = true;
exit(0); exit(0);
}else if(metaValue == "debug:record") { }
try{ else if(metaValue == "debug:record") {
try {
json body = json::parse(body2str(v[2])); json body = json::parse(body2str(v[2]));
if(body.count("data") != 0 && body["data"].is_object() && body["data"].count("start") != 0 && body["data"]["start"].is_number() && body["data"].count("end") != 0 && body["data"]["end"].is_number()) { if(body.count("data") != 0 && body["data"].is_object() && body["data"].count("start") != 0 && body["data"]["start"].is_number() && body["data"].count("end") != 0 && body["data"]["end"].is_number()) {
json evt; json evt;
...@@ -180,26 +182,31 @@ private: ...@@ -180,26 +182,31 @@ private:
lock_guard<mutex> lock(this->mutEvent); lock_guard<mutex> lock(this->mutEvent);
eventQueue.push(evt.dump()); eventQueue.push(evt.dump());
cvEvent.notify_one(); cvEvent.notify_one();
} }
bProcessed = true; bProcessed = true;
} }
}catch(exception &e) { }
catch(exception &e) {
spdlog::error("evslicer {} exception in handleCloudMsg: {}", selfId, e.what()); spdlog::error("evslicer {} exception in handleCloudMsg: {}", selfId, e.what());
} }
}else if(metaValue == "debug:xxx"){ }
else if(metaValue == "debug:xxx") {
// TODO: remove debug feature // TODO: remove debug feature
bProcessed = true; bProcessed = true;
}else if(metaValue == "debug:list_files"){ }
else if(metaValue == "debug:list_files") {
// TODO: remove debug feature // TODO: remove debug feature
printVideoFiles(this->sTsList); printVideoFiles(this->sTsList);
bProcessed = true; bProcessed = true;
}else if(metaValue == "debug:toggle_log") { }
else if(metaValue == "debug:toggle_log") {
// TODO: remove debug feature // TODO: remove debug feature
static bool toggle = false; static bool toggle = false;
toggle = !toggle; toggle = !toggle;
if(toggle){ if(toggle) {
spdlog::set_level(spdlog::level::debug); spdlog::set_level(spdlog::level::debug);
}else{ }
else {
spdlog::set_level(spdlog::level::info); spdlog::set_level(spdlog::level::info);
} }
...@@ -309,14 +316,16 @@ private: ...@@ -309,14 +316,16 @@ private:
int portPub = 5556; int portPub = 5556;
if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) { if(evpuller.count("portPub") != 0 && evpuller["portPub"].is_number_integer()) {
portPub = evpuller["portPub"]; portPub = evpuller["portPub"];
}else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()){ }
else if(evpuller.count("port-pub") != 0 && evpuller["port-pub"].is_number_integer()) {
portPub = evpuller["port-pub"]; portPub = evpuller["port-pub"];
} }
int portRouter = 5550; int portRouter = 5550;
if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) { if(evmgr.count("portRouter") != 0 && evmgr["portRouter"].is_number_integer()) {
portRouter = evmgr["portRouter"]; portRouter = evmgr["portRouter"];
}else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) { }
else if(evmgr.count("port-router") != 0 && evmgr["port-router"].is_number_integer()) {
portRouter = evmgr["port-router"]; portRouter = evmgr["port-router"];
} }
...@@ -698,7 +707,8 @@ protected: ...@@ -698,7 +707,8 @@ protected:
} }
} }
void insertTsList(set<long> &_list, long elem, int maxSize) { void insertTsList(set<long> &_list, long elem, int maxSize)
{
// _list.insert(lower_bound(_list.begin(), _list.end(), elem), elem); // _list.insert(lower_bound(_list.begin(), _list.end(), elem), elem);
if(_list.size() == 0) { if(_list.size() == 0) {
_list.insert(_list.begin(),elem); _list.insert(_list.begin(),elem);
...@@ -708,14 +718,15 @@ protected: ...@@ -708,14 +718,15 @@ protected:
auto itr = _list.rbegin(); auto itr = _list.rbegin();
for(; itr != _list.rend(); itr++) { for(; itr != _list.rend(); itr++) {
if(*itr < elem){ if(*itr < elem) {
break; break;
} }
} }
if(itr == _list.rbegin() ) { if(itr == _list.rbegin() ) {
_list.insert(_list.end(), elem); _list.insert(_list.end(), elem);
}else{ }
else {
_list.insert(itr.base(), elem); _list.insert(itr.base(), elem);
} }
...@@ -731,7 +742,7 @@ protected: ...@@ -731,7 +742,7 @@ protected:
// file monitor callback // file monitor callback
static void fileMonHandler(const std::vector<event>& evts, void *pUserData) static void fileMonHandler(const std::vector<event>& evts, void *pUserData)
{ {
static string lastFile; static string lastFile;
string ext = ".mp4"; string ext = ".mp4";
auto self = static_cast<EvSlicer*>(pUserData); auto self = static_cast<EvSlicer*>(pUserData);
...@@ -755,7 +766,8 @@ protected: ...@@ -755,7 +766,8 @@ protected:
auto ts = self->videoFileName2Ts(baseName); auto ts = self->videoFileName2Ts(baseName);
if(ts == -1) { if(ts == -1) {
spdlog::error("evslicer {} fileMonHandler failed to process file: {}", self->selfId, lastFile); spdlog::error("evslicer {} fileMonHandler failed to process file: {}", self->selfId, lastFile);
}else{ }
else {
self->insertTsList(self->sTsList, ts, self->numSlices); self->insertTsList(self->sTsList, ts, self->numSlices);
} }
} }
...@@ -774,7 +786,7 @@ protected: ...@@ -774,7 +786,7 @@ protected:
vector<string> findSlicesByRange(long tss, long tse, int offsetS, int offsetE) vector<string> findSlicesByRange(long tss, long tse, int offsetS, int offsetE)
{ {
vector<string> ret; vector<string> ret;
lock_guard<mutex> lg(mutTsList); lock_guard<mutex> lg(mutTsList);
if(this->sTsList.size() == 0) { if(this->sTsList.size() == 0) {
return ret; return ret;
...@@ -788,7 +800,7 @@ protected: ...@@ -788,7 +800,7 @@ protected:
spdlog::info("evslicer {} event range ({}, {}) is not in range ({}, {}).", selfId, tss, tse, first, end); spdlog::info("evslicer {} event range ({}, {}) is not in range ({}, {}).", selfId, tss, tse, first, end);
return ret; return ret;
} }
first = end = 0; first = end = 0;
set<long> tmp; set<long> tmp;
int found = 0; int found = 0;
...@@ -920,7 +932,7 @@ public: ...@@ -920,7 +932,7 @@ public:
this->eventQueue.pop(); this->eventQueue.pop();
} }
} }
if(evt.empty()) { if(evt.empty()) {
continue; continue;
} }
...@@ -947,9 +959,10 @@ public: ...@@ -947,9 +959,10 @@ public:
if(tse < first) { if(tse < first) {
spdlog::info("evslicer {} thEventHandler event range ({}, {}) is not in range ({}, {}).", selfId, tss, tse, first, end); spdlog::info("evslicer {} thEventHandler event range ({}, {}) is not in range ({}, {}).", selfId, tss, tse, first, end);
continue; continue;
}else if(first == 0||tse > end) { }
else if(first == 0||tse > end) {
spdlog::info("evslicer {} thEventHandler event range ({}, {}) is not in range ({}, {}), resched to run in {}s.", selfId, tss, tse, first, end, this->seconds + 5); spdlog::info("evslicer {} thEventHandler event range ({}, {}) is not in range ({}, {}), resched to run in {}s.", selfId, tss, tse, first, end, this->seconds + 5);
auto th = thread([evt, this](){ auto th = thread([evt, this]() {
this_thread::sleep_for(chrono::seconds(this->seconds + 5)); this_thread::sleep_for(chrono::seconds(this->seconds + 5));
lock_guard<mutex> lock(this->mutEvent); lock_guard<mutex> lock(this->mutEvent);
this->eventQueue.push(evt); this->eventQueue.push(evt);
...@@ -987,6 +1000,7 @@ public: ...@@ -987,6 +1000,7 @@ public:
fileNames.push_back(fname); fileNames.push_back(fname);
sf+="\tfile\t" + fname + "\n"; sf+="\tfile\t" + fname + "\n";
} }
if(hasError) { if(hasError) {
continue; continue;
} }
...@@ -1010,18 +1024,19 @@ public: ...@@ -1010,18 +1024,19 @@ public:
postArgs["params"] = params; postArgs["params"] = params;
postArgs["fileNames"] = fileNames; postArgs["fileNames"] = fileNames;
string fname = dirDest + params["startTime"].get<string>() + "_" + params["endTime"].get<string>() + "evt.json"; string fname = dirDest + params["startTime"].get<string>() + "_" + params["endTime"].get<string>() + "evt.json";
try{ try {
ofstream ofs(fname); ofstream ofs(fname);
ofs << postArgs; ofs << postArgs;
for(auto &f:fileNames){ for(auto &f:fileNames) {
fs::copy(fs::path(string(f)),fs::path(dirDest)); fs::copy(fs::path(string(f)),fs::path(dirDest));
} }
}
}catch(exception &e) { catch(exception &e) {
spdlog::error("evcloudsvc {} {}:{} exception: {}", selfId, __FILE__, __LINE__, e.what()); spdlog::error("evcloudsvc {} {}:{} exception: {}", selfId, __FILE__, __LINE__, e.what());
} }
}else{ }
else {
spdlog::info("evslicer {} retrying upload", selfId); spdlog::info("evslicer {} retrying upload", selfId);
jEvt["cnt"] = jEvt["cnt"].get<int>() - 1; jEvt["cnt"] = jEvt["cnt"].get<int>() - 1;
lock_guard<mutex> lock(this->mutEvent); lock_guard<mutex> lock(this->mutEvent);
...@@ -1030,24 +1045,26 @@ public: ...@@ -1030,24 +1045,26 @@ public:
eventQueue.pop(); eventQueue.pop();
} }
cvEvent.notify_one(); cvEvent.notify_one();
} }
} }
} }
else { else {
spdlog::info("evslicer {} upload ({}, {}). local({}, {}). resp: {} files:\n{}", selfId, tss, tse, first, end, strResp, sf); spdlog::info("evslicer {} upload ({}, {}). local({}, {}). resp: {} files:\n{}", selfId, tss, tse, first, end, strResp, sf);
if(ret > 0){ if(ret > 0) {
try{ try {
auto resp = json::parse(strResp); auto resp = json::parse(strResp);
//TODO: open this swith when video server has implemented this functionality //TODO: open this swith when video server has implemented this functionality
if(true){ if(true) {
if(resp.count("code") != 0 && resp["code"] != 0) { if(resp.count("code") != 0 && resp["code"] != 0) {
if(resp["code"] == 4|| resp["code"] == 7) { if(resp["code"] == 4|| resp["code"] == 7) {
if(jEvt.count("cnt") == 0) { if(jEvt.count("cnt") == 0) {
jEvt["cnt"] = 2; jEvt["cnt"] = 2;
}else{ }
else {
if(jEvt["cnt"].get<int>() <= 0) { if(jEvt["cnt"].get<int>() <= 0) {
spdlog::error("evslicer {} failed to upload videos over N times, abort retrying: {}", this->selfId, evt); spdlog::error("evslicer {} failed to upload videos over N times, abort retrying: {}", this->selfId, evt);
}else{ }
else {
jEvt["cnt"] = jEvt["cnt"].get<int>() - 1; jEvt["cnt"] = jEvt["cnt"].get<int>() - 1;
lock_guard<mutex> lock(this->mutEvent); lock_guard<mutex> lock(this->mutEvent);
this->eventQueue.push(jEvt.dump()); this->eventQueue.push(jEvt.dump());
...@@ -1057,15 +1074,18 @@ public: ...@@ -1057,15 +1074,18 @@ public:
cvEvent.notify_one(); cvEvent.notify_one();
} }
} }
}else if(resp["code"] == 6) { }
else if(resp["code"] == 6) {
// TODO: cloud storage issue. need stratigy policy // TODO: cloud storage issue. need stratigy policy
spdlog::warn("evslicer {} TODO: handle cloud storage", this->selfId); spdlog::warn("evslicer {} TODO: handle cloud storage", this->selfId);
}else{ }
else {
spdlog::error("evslicer {} failed to upload videos. abort retry.", this->selfId); spdlog::error("evslicer {} failed to upload videos. abort retry.", this->selfId);
} }
} }
} }
}catch(exception &e) { }
catch(exception &e) {
spdlog::error("evslicer {} {}:{} exception: {}", this->selfId, __FILE__, __LINE__, e.what()); spdlog::error("evslicer {} {}:{} exception: {}", this->selfId, __FILE__, __LINE__, e.what());
} }
} }
......
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论