Skip to content

E
evcamera
提交 b92bb722 authored 作者: blu's avatar blu
浏览文件
操作

recording & refactoring

上级 2c0181b7
显示空白字符变更
内嵌 并排
正在显示 包含 3475 行增加 和 611 行删除
hi3518/CMakeCache.txt
浏览文件 @ b92bb722
......@@ -185,12 +185,18 @@ evcamera_BINARY_DIR:STATIC=/mnt/shared/evcamera/hi3518
//Value Computed by CMake
evcamera_SOURCE_DIR:STATIC=/mnt/shared/evcamera/hi3518
//Dependencies for target
evutils_LIB_DEPENDS:STATIC=
//Dependencies for target
mime_LIB_DEPENDS:STATIC=
//Dependencies for target
motion_LIB_DEPENDS:STATIC=
//Dependencies for target
mqtthelper_LIB_DEPENDS:STATIC=
//Dependencies for target
ntp_LIB_DEPENDS:STATIC=
......
hi3518/CMakeLists.txt
浏览文件 @ b92bb722
......@@ -20,15 +20,11 @@ link_directories(${COMMON_LIB_DIR} /root/xiongmai/arm-himix100-linux/target/lib/
#add_library(mime STATIC mime.c)
add_library(ntp STATIC ntp.cc)
add_library(tcp_client STATIC raw_tcp.cc)
add_library(smart STATIC smart.cc)
add_library(evutils STATIC evutils.cc)
add_library(mqtthelper STATIC ../include/mqtt_helper.cc)
add_library(motion STATIC motion.cc)
add_library(smart STATIC smart.cc)
#add_library(util STATIC utils.cpp)
add_executable(test_mqtt test_mqtt.cc)
target_link_libraries(test_mqtt PUBLIC paho-mqtt3a fmt m dl pthread)
add_executable(evcamera main.cc)
target_link_libraries(evcamera PUBLIC ntp paho-mqtt3a tcp_client smart motion zmq ${COMMON_LIBS} ${XM_LIBS})
add_executable(ntp_client ntp_main.cc)
target_link_libraries(ntp_client PUBLIC ntp)
target_link_libraries(evcamera PUBLIC ntp paho-mqtt3a mqtthelper tcp_client evutils smart motion zmq ${COMMON_LIBS} ${XM_LIBS})
hi3518/utils.hpp → hi3518/evutils.cc
浏览文件 @ b92bb722
#ifndef __EV_UTILS_HPP__
#define __EV_UTILS_HPP__
#include <string>
#include <jsoncons/json.hpp>
#include <fstream>
#include <algorithm>
#include <sys/statvfs.h>
#include <mutex>
#include <set>
#include <zmq.h>
#include <string>
#include <spdlog/spdlog.h>
#include <fmt/format.h>
#include "evutils.hpp"
using namespace std;
using namespace jsoncons;
using namespace jsoncons::literals;
namespace evutils {
const char version[] = "EVC20200429";
/// for mqtt proto: json payload
const string kMsgCmd = "cmd";
const string kMsgCmdConfig = "config";
const string kMsgConfigVgw = "vgw";
const string kMsgConfigMqtt = "mqtt";
const string kMsgConfigUpload = "upload";
const string kMsgConfigFeatures = "features";
const string kMsgConfigEnable = "enable";
const string kMsgConfigLevel = "level";
const string kMsgConfigInterval = "interval";
const string kMsgConfigDuration = "duration";
const string kMsgConfigHumanThreash = "humanThreash";
const string kMsgConfigFaceThreash = "faceThresh";
const string kMsgConfigRegion = "region";
const string kMsgConfigMotion = "motion";
const string kMsgConfigRecord = "record";
const string kMsgConfigAi = "ai";
const string kMsgCmdUploadVideo = "upload_video";
const string kMsgCmdStatus = "status";
const string kMsgCmdGetConfig = "get_config";
const string kMsgCmdReboot = "reboot";
const string kMsgCmdPtz = "ptz";
const string kMsgType = "type";
const string kMsgCata = "cata";
const string kMsgTypeReport = "report";
const string kMsgTypeResponse = "response";
const string kMsgTime = "time";
const string kMsgData = "data";
const string kMsgRid = "rid";
const string kMsgCode = "code";
const string kMsgMsg = "msg";
const string kMsgSn = "sn";
/// default video gateway info
char hostArr[] = "192.168.55.104";
char portArr[] = "7123";
char *host = hostArr, *port = portArr;
string recFilePath = "/mnt/sd/records/";
const long long TS_2020 = 1577836800000L;
const char consts::version[] = "EVC20200429";
const string consts::kMsgCmd = "cmd";
const string consts::kMsgCmdConfig = "config";
const string consts::kMsgConfigVgw = "vgw";
const string consts::kMsgConfigMqtt = "mqtt";
const string consts::kMsgConfigUpload = "upload";
const string consts::kMsgConfigFeatures = "features";
const string consts::kMsgConfigEnable = "enable";
const string consts::kMsgConfigLevel = "level";
const string consts::kMsgConfigInterval = "interval";
const string consts::kMsgConfigDuration = "duration";
const string consts::kMsgConfigHumanThreash = "humanThreash";
const string consts::kMsgConfigFaceThreash = "faceThresh";
const string consts::kMsgConfigRegion = "region";
const string consts::kMsgConfigMotion = "motion";
const string consts::kMsgConfigRecord = "record";
const string consts::kMsgConfigAi = "ai";
const string consts::kMsgCmdUploadVideo = "upload_video";
const string consts::kMsgCmdStatus = "status";
const string consts::kMsgCmdGetConfig = "get_config";
const string consts::kMsgCmdReboot = "reboot";
const string consts::kMsgCmdPtz = "ptz";
const string consts::kMsgType = "type";
const string consts::kMsgCata = "category";
const string consts::kMsgTypeReport = "report";
const string consts::kMsgTypeResponse = "response";
const string consts::kMsgTime = "time";
const string consts::kMsgData = "data";
const string consts::kMsgRid = "rid";
const string consts::kMsgCode = "code";
const string consts::kMsgMsg = "msg";
const string consts::kMsgSn = "sn";
const string consts::recFilePath = "/mnt/sd/records/";
const long long consts::TS_2020 = 1577836800000L;
/// topics
string sub_topic = "evcamera/v1.0/request/";
string pub_topic = "evcamera/v1.0/res_rep";
string mqtt_url = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
const string consts::sub_topic = "evcamera/v1.0/request/";
const string consts::pub_topic_response = "evcamera/v1.0/response/";
const string consts::pub_topic_report = "evcamera/v1.0/report";
const string consts::mqtt_url = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
consts &consts::self(){
static consts cons;
return cons;
}
inline string msg_field(const json &msg, const string &field)
string msg_field(const json &msg, const string &field)
{
return msg.get_value_or<string>(field, "");
}
bool is_sdcard_avail(char *path = nullptr)
bool is_sdcard_avail(char *path)
{
char p[] = "/sys/block/mmcblk0/size";
if(path == nullptr){
......@@ -117,7 +101,7 @@ json make_default_config(){
)"_json;
}
void get_mac_addr(char *buf, char *intf = nullptr){
void get_mac_addr(char *buf, char *intf){
char p[] = "/sys/class/net/eth0/address";
if(intf == nullptr){
intf = p;
......@@ -135,7 +119,7 @@ void get_mac_addr(char *buf, char *intf = nullptr){
*(buf+i) = 0;
}
bool get_sdcard_megabytes(uint64_t &total, uint64_t &free, char* path = nullptr)
bool get_sdcard_megabytes(uint64_t &total, uint64_t &free, char* path )
{
struct statvfs stat;
char p[] = "/mnt/sd";
......@@ -155,133 +139,8 @@ bool get_sdcard_megabytes(uint64_t &total, uint64_t &free, char* path = nullptr)
return true;
}
template <typename T>
using cb_remove_elem = void(*)(T elem);
template <typename TN>
class OrderedList {
private:
set<TN> list_;
size_t maxSize;
mutex mut;
TN oldestTs;
cb_remove_elem<TN> fn_remove;
unsigned long cntInsert = 0;
public:
OrderedList() = delete;
OrderedList(ssize_t maxSize, cb_remove_elem<TN> fn_remove):maxSize(maxSize), fn_remove(fn_remove){}
void insert(TN elem, cb_remove_elem<TN> fn=nullptr)
{
// list_.insert(lower_bound(list_.begin(), list_.end(), elem), elem);
if(cntInsert % 10 == 0) {
oldestTs = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count() - 60*2*1000*maxSize;
spdlog::info("record: maxSlices {}, oldestTs {}", maxSize, oldestTs);
}
++cntInsert;
if(elem < oldestTs){
if(fn != nullptr){
(*fn)(elem);
}else if(fn_remove != nullptr){
(*fn_remove)(elem);
}
return;
}
if(list_.size() == 0) {
list_.insert(list_.begin(),elem);
return;
}
auto itr = list_.rbegin();
for(; itr != list_.rend(); itr++) {
if(*itr < elem) {
break;
}
}
if(itr == list_.rbegin() ) {
list_.insert(list_.end(), elem);
}
else {
list_.insert(itr.base(), elem);
}
if(list_.size() > maxSize) {
lock_guard<mutex> lg(mut);
auto ts = *(list_.begin());
list_.erase(list_.begin());
if(fn != nullptr){
(*fn)(ts);
}else if(fn_remove != nullptr){
(*fn_remove)(ts);
}else{
// no callback available
}
// auto baseName = videoFileTs2Name(ts);
// fs::path fname(this->urlOut + "/" + baseName + ".mp4");
// fs::remove(fname);
}
}
set<TN> findByRange(TN tss, TN tse, TN & offsetS, TN &offsetE){
set<TN> ret;
lock_guard<mutex> lg(mut);
if(list_.size() == 0) {
return ret;
}
TN first = *(list_.begin());
auto _it = list_.end();
TN end = *(--_it);
if(tse < first||tss > end) {
spdlog::info("range requested ({}, {}) is not in range existed ({}, {}).", tss, tse, first, end);
return ret;
}
first = end = 0;
int found = 0;
TN last = 0;
auto itr = list_.rbegin();
for(; itr != list_.rend(); itr++) {
if(*itr > tse) {
continue;
}
if(*itr <= tse) {
if(found != 1) {
spdlog::info("\t matched : {}, s:{}, e:{}", *itr, tss, tse);
found = 1;
// check the end offset, not guaranteed
if( itr != list_.rbegin()){
auto t = itr;
last = *(--t);
}
}
ret.insert(*itr);
if(tss >= *itr) {
break;
}
}
}
if(found == 1) {
auto itr = ret.begin();
offsetS = tss - *itr;
if(last != 0) {
offsetE = last - tse;
}
}
return ret;
}
};
/// returns negtive for failure, otherwise success
int setupDealer(void **ctx, void **s, string ident, string addr, int sndQS=0, int timeoutMs = -1) {
int setupDealer(void **ctx, void **s, string ident, string addr, int sndQS, int timeoutMs) {
int ret = 0;
*ctx = zmq_ctx_new();
*s = zmq_socket(*ctx, ZMQ_DEALER);
......@@ -319,7 +178,7 @@ int setupDealer(void **ctx, void **s, string ident, string addr, int sndQS=0, in
return ret;
}
int setupRouter(void **ctx, void **s, string addr, int rcvQS=0){
int setupRouter(void **ctx, void **s, string addr, int rcvQS){
int ret = 0;
*ctx = zmq_ctx_new();
*s = zmq_socket(*ctx, ZMQ_ROUTER);
......@@ -379,5 +238,3 @@ int z_recv_multiple(void *s, vector<uint8_t> &buf, int &frames) {
}
\ No newline at end of file
#endif
\ No newline at end of file
hi3518/evutils.hpp 0 → 100644
浏览文件 @ b92bb722
#ifndef __EV_UTILS_HPP__
#define __EV_UTILS_HPP__
#pragma once
#include <string>
#include <jsoncons/json.hpp>
#include <fstream>
#include <algorithm>
#include <sys/statvfs.h>
#include <mutex>
#include <set>
#include <zmq.h>
#include <string>
#include <spdlog/spdlog.h>
#include <fmt/format.h>
#include <jsoncons/json.hpp>
namespace evutils {
using namespace std;
using namespace jsoncons;
typedef struct consts{
/// for mqtt proto: json payload
static const char version[];
static const string kMsgCmd;
static const string kMsgCmdConfig;
static const string kMsgConfigVgw;
static const string kMsgConfigMqtt;
static const string kMsgConfigUpload;
static const string kMsgConfigFeatures;
static const string kMsgConfigEnable;
static const string kMsgConfigLevel;
static const string kMsgConfigInterval;
static const string kMsgConfigDuration;
static const string kMsgConfigHumanThreash;
static const string kMsgConfigFaceThreash;
static const string kMsgConfigRegion;
static const string kMsgConfigMotion;
static const string kMsgConfigRecord;
static const string kMsgConfigAi;
static const string kMsgCmdUploadVideo;
static const string kMsgCmdStatus;
static const string kMsgCmdGetConfig;
static const string kMsgCmdReboot;
static const string kMsgCmdPtz;
static const string kMsgType;
static const string kMsgCata;
static const string kMsgTypeReport;
static const string kMsgTypeResponse;
static const string kMsgTime;
static const string kMsgData;
static const string kMsgRid;
static const string kMsgCode;
static const string kMsgMsg;
static const string kMsgSn;
static const string recFilePath;
static const long long TS_2020;
/// topics
static const string sub_topic;
static const string pub_topic_response;
static const string pub_topic_report;
static const string mqtt_url;
static consts& self();
}consts;
extern string msg_field(const json &msg, const string &field);
extern bool is_sdcard_avail(char *path = nullptr);
extern json make_default_config();
extern void get_mac_addr(char *buf, char *intf = nullptr);
extern bool get_sdcard_megabytes(uint64_t &total, uint64_t &free, char* path = nullptr);
extern int setupDealer(void **ctx, void **s, string ident, string addr, int sndQS=0, int timeoutMs = -1);
extern int setupRouter(void **ctx, void **s, string addr, int rcvQS=0);
extern int z_recv_multiple(void *s, vector<uint8_t> &buf, int &frames);
template <typename T>
using cb_remove_elem = void(*)(T elem);
template <typename TN>
class OrderedList {
private:
set<TN> list_;
size_t maxSize;
mutex mut;
TN oldestTs;
cb_remove_elem<TN> fn_remove;
unsigned long cntInsert = 0;
public:
OrderedList() = delete;
OrderedList(ssize_t maxSize, cb_remove_elem<TN> fn_remove):maxSize(maxSize), fn_remove(fn_remove) {}
void insert(TN elem, cb_remove_elem<TN> fn=nullptr)
{
// list_.insert(lower_bound(list_.begin(), list_.end(), elem), elem);
if(cntInsert % 10 == 0) {
oldestTs = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count() - 60*2*1000*maxSize;
spdlog::info("record: maxSlices {}, oldestTs {}", maxSize, oldestTs);
}
++cntInsert;
if(elem < oldestTs) {
if(fn != nullptr) {
(*fn)(elem);
}
else if(fn_remove != nullptr) {
(*fn_remove)(elem);
}
return;
}
if(list_.size() == 0) {
list_.insert(list_.begin(),elem);
return;
}
auto itr = list_.rbegin();
for(; itr != list_.rend(); itr++) {
if(*itr < elem) {
break;
}
}
if(itr == list_.rbegin() ) {
list_.insert(list_.end(), elem);
}
else {
list_.insert(itr.base(), elem);
}
if(list_.size() > maxSize) {
lock_guard<mutex> lg(mut);
auto ts = *(list_.begin());
list_.erase(list_.begin());
if(fn != nullptr) {
(*fn)(ts);
}
else if(fn_remove != nullptr) {
(*fn_remove)(ts);
}
else {
// no callback available
}
}
}
set<TN> findByRange(TN tss, TN tse, TN & offsetS, TN &offsetE)
{
set<TN> ret;
lock_guard<mutex> lg(mut);
if(list_.size() == 0) {
return ret;
}
TN first = *(list_.begin());
auto _it = list_.end();
TN end = *(--_it);
if(tse < first||tss > end) {
spdlog::info("range requested ({}, {}) is not in range existed ({}, {}).", tss, tse, first, end);
return ret;
}
first = end = 0;
int found = 0;
TN last = 0;
auto itr = list_.rbegin();
for(; itr != list_.rend(); itr++) {
if(*itr > tse) {
continue;
}
if(*itr <= tse) {
if(found != 1) {
spdlog::info("\t matched : {}, s:{}, e:{}", *itr, tss, tse);
found = 1;
// check the end offset, not guaranteed
if( itr != list_.rbegin()) {
auto t = itr;
last = *(--t);
}
}
ret.insert(*itr);
if(tss >= *itr) {
break;
}
}
}
if(found == 1) {
auto itr = ret.begin();
offsetS = tss - *itr;
if(last != 0) {
offsetE = last - tse;
}
}
return ret;
}
set<TN> & items()
{
return list_;
}
};
}
#endif
\ No newline at end of file
hi3518/main.cc
浏览文件 @ b92bb722
......@@ -16,15 +16,16 @@
#include "smart.h"
#include "motion.h"
#include <thread>
#include "mqttmgr.hpp"
#include "utils.hpp"
#include "evutils.hpp"
#include <ghc/filesystem.hpp>
#include <mqtt_helper.hpp>
#include <signal.h>
using namespace std;
using namespace jsoncons;
using namespace evutils;
using namespace chrono_literals;
namespace fs = ghc::filesystem;
using namespace evutils;
///
#define NUM_IFRAME_PICK 2
......@@ -33,51 +34,58 @@ namespace fs = ghc::filesystem;
#define INTERVAL_LOG_QFULL 3 // its about 60/15 *3 secs
/// device info
static char dev_sn[TERMINAL_SN_SIZE] = {0};
char dev_sn[TERMINAL_SN_SIZE] = {0};
/// default video gateway info
static int videoQuality = MAQUE_IMG_QUALITY_BETTER;
static uint64_t frameCntTotal = 0;
static uint32_t frameCntIframe = 0;
static uint32_t frameCntPframe = 0;
static uint64_t packetId = 0;
static spdlog::level::level_enum logLevel = spdlog::level::info;
static bool bPFrameAvail = false;
static bool bNeedRetry = false;
static int readFaildCnt = 0;
static bool enablePush = false;
static bool enableRecord = false;
static bool bCanRecord = false;
static bool bConnected = false;
static queue<DataItem> _frameQueue;
static mutex _mutFrame;
static condition_variable _condFrame;
static Notifier _notiFrame = {&_mutFrame, &_condFrame};
static CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
static bool bGotTime = false;
int videoQuality = MAQUE_IMG_QUALITY_BETTER;
uint64_t frameCntTotal = 0;
uint32_t frameCntIframe = 0;
uint32_t frameCntPframe = 0;
uint64_t packetId = 0;
spdlog::level::level_enum logLevel = spdlog::level::info;
bool bPFrameAvail = false;
bool bNeedRetry = false;
int readFaildCnt = 0;
bool enablePush = false;
bool enableRecord = false;
bool bCanRecord = false;
bool bConnected = false;
queue<DataItem> _frameQueue;
mutex _mutFrame;
condition_variable _condFrame;
Notifier _notiFrame = {&_mutFrame, &_condFrame};
CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
bool bGotTime = false;
// mqtt
MqttHelper *gMqttClient = nullptr;
// zmq
static void *pPubCtx = nullptr;
static void *pPub = nullptr;
static const string strPubUrl = "inproc://frame";
static void *pSub = nullptr;
static void *pDealerCtx = nullptr;
static void *pDealer = nullptr;
static string vgwUrl = "tcp://192.168.55.104:7123";
void *pPubCtx = nullptr;
void *pPub = nullptr;
const string strPubUrl = "inproc://frame";
void *pSub = nullptr;
void *pDealerCtx = nullptr;
void *pDealer = nullptr;
string vgwUrl = "tcp://192.168.55.104:7123";
//
static string configFilePath = "/etc/evcamera.json";
static json gJsonConfig;
string configFilePath = "/etc/evcamera.json";
json gJsonConfig;
char hostArr[] = "192.168.55.104";
char portArr[] = "7123";
char *host = hostArr;
char *port = portArr;
//
static OrderedList<long long> *gRecFilesList = nullptr;
OrderedList<long long> *gRecFilesList = nullptr;
//
static string strMaQuePath = "/mnt/sd/Config/";
string strMaQuePath = "/mnt/sd/Config/";
static string strSDCardReservedPaths[] = {recFilePath, strMaQuePath};
/// why using singleton ? because global vriable initialization order between files is undefined!
/// this avoids getting empty result
string strSDCardReservedPaths[] = {consts::self().recFilePath, strMaQuePath};
/// flags used to stop modules
/// 0: stopped, 1: requested to stop, 2: runningstatic int
......@@ -131,7 +139,6 @@ void clean_up(int sig)
fclose(args.recFD);
}
// while (args.dataq->size() > 0) {
// DataItem elem = args.dataq->front();
// args.dataq->pop();
......@@ -344,20 +351,34 @@ void frame_send_entry(void *args)
}
}
void print_ts_files(OrderedList<long long> &_list) {
spdlog::info("print file ts {}:", _list.items().size());
int cnt = 0;
printf("\t");
for(auto &k:_list.items()){
cnt++;
printf("%lld, ", k);
if(cnt %6 == 0){
printf("\n\t");
}
}
spdlog::info("\nlist_videos done!");
}
void load_sd_files(OrderedList<long long> &_list){
auto now = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
try {
string fname, baseName;
for (const auto & entry : fs::directory_iterator(recFilePath)) {
for (const auto & entry : fs::directory_iterator(consts::recFilePath)) {
fname = entry.path().c_str();
if(entry.file_size() == 0 || !entry.is_regular_file()||entry.path().extension() != ".h264") {
spdlog::warn("{} loadVideoFiles skipped {} (empty/directory/!h264)", entry.path().c_str());
spdlog::warn("load_sd_files skipped {} (empty/directory/!h264)", entry.path().c_str());
continue;
}
auto ts = stoi(entry.path().stem().c_str());
if(ts < TS_2020){
long long ts = stoll(entry.path().stem().string());
if(ts < consts::TS_2020){
/// TODO: files records when offline with no valid time got
fs::remove(entry.path());
spdlog::warn("remove records having invalid timestamp: {}", ts);
......@@ -397,10 +418,9 @@ void record_video_entry(void *args)
// calc num of slices
// reserve 0.5GB space
ssize_t maxSlices = (total - 512)/10;
OrderedList<long long> tsList(maxSlices, remove_ts_file);
gRecFilesList = &tsList;
gRecFilesList = new OrderedList<long long>(maxSlices, remove_ts_file);
// load existing videos
load_sd_files(tsList);
load_sd_files(*gRecFilesList);
// setup sub
void *pSub = zmq_socket(pPubCtx, ZMQ_SUB);
......@@ -441,13 +461,13 @@ void record_video_entry(void *args)
fp->close();
delete fp;
}
fp = new fstream(recFilePath+to_string(nowTs), std::ios::out|std::ios::binary);
fp = new fstream(consts::recFilePath+to_string(nowTs) + ".h264", std::ios::out|std::ios::binary);
if(!fp->is_open()){
spdlog::error("failed to open record file");
delete fp;
fp = nullptr;
}else{
tsList.insert(nowTs);
gRecFilesList->insert(nowTs);
}
preTs = nowTs;
}
......@@ -470,7 +490,7 @@ void record_video_entry(void *args)
string verify_request(json &js)
{
string ret;
string check[] = {kMsgCmd,kMsgData, kMsgRid, kMsgTime, kMsgRid};
string check[] = {consts::kMsgCmd, consts::kMsgData, consts::kMsgRid, consts::kMsgTime};
for(auto &k: check) {
if(!js.contains(k)) {
ret += k + ",";
......@@ -484,76 +504,10 @@ string verify_request(json &js)
return ret;
}
/// low level api for report and response
void _report_response(string topic, string message)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
}
auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
if(rc < 0) {
spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
}
else {
spdlog::info("successfully pub message at {}:{}", topic, message);
}
}
/// having data field along with code and msg
/// signature {"rid", "code", "msg", "cata", "data":{}}
void report_response_json(string topic, json& message)
{
// time
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
message.insert_or_assign(kMsgTime, ts);
// rid
auto rid = message["rid"].as<string>();
message.insert_or_assign(kMsgRid, rid);
// type
if(rid.empty()) {
message.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
message.insert_or_assign(kMsgType, kMsgTypeResponse);
}
// sn
message.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, message.to_string());
}
/// no data field. only code and msg field
void report_response_args(string topic, int code, string message, string cata, string rid, json &&data)
{
json js;
/// current timestamp
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
js.insert_or_assign(kMsgTime, ts);
/// rid
js.insert_or_assign(kMsgRid, rid);
/// type
if(rid.empty()) {
js.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
js.insert_or_assign(kMsgType, kMsgTypeResponse);
}
js.insert_or_assign(kMsgCata, cata);
///
js.insert_or_assign(kMsgCode, code);
js.insert_or_assign(kMsgMsg, message);
js.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, js.to_string());
}
string verify_config(json &data)
{
string str, strWarn;
string veri[] = {kMsgConfigVgw, kMsgConfigMqtt, kMsgConfigUpload};
string veri[] = {consts::kMsgConfigVgw, consts::kMsgConfigMqtt, consts::kMsgConfigUpload};
for(auto &k:veri) {
auto value = msg_field(data, k);
if(value.empty()) {
......@@ -591,59 +545,97 @@ string verify_config(json &data)
/// handles all incoming request from cloud services
void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
{
char msg[len];
char msg[len+1] = {0};
memcpy(msg, data, len);
spdlog::info("received request on {}:\n{}", topic, (char*)msg);
try {
auto js = json::parse(msg);
spdlog::info("received request on {}:\n{}", topic, (char*)msg);
auto str = verify_request(js);
if(!str.empty()) {
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(kMsgCmd)?js[kMsgCmd].as<string>():string(""), js.contains(kMsgRid)?js[kMsgRid].as<string>():string(""), json());
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), json());
return;
}
/// dispatch msg
auto cmd = js[kMsgCmd].as<string>();
if(cmd == kMsgCmdConfig) {
auto cmd = js[consts::kMsgCmd].as<string>();
auto rid = js[consts::kMsgRid].as<string>();
auto data = js[consts::kMsgData].as<json>();
/// TODO: implements cmds
if(cmd == consts::kMsgCmdConfig) {
// apply config
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}
else if(cmd == kMsgCmdUploadVideo) {
else if(cmd == consts::kMsgCmdUploadVideo) {
// upload video
if(gRecFilesList == nullptr){
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, -1, "no sd card", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}else{
auto tss = data["start"].as<long long>();
auto tse = data["end"].as<long long>();
// guess unit in ms or s
if(tss/100000000000 == 0){
tss = tss * 1000;
tse = tse * 1000;
}
long long offsetE = 0, offsetS = 0;
auto res = gRecFilesList->findByRange(tss, tse, offsetS, offsetE);
if(res.empty()){
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, -2, "no video found", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}else{
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", js.contains(consts::consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
print_ts_files(*gRecFilesList);
printf("matched %lld, %lld:\n\t", offsetS, offsetE);
for(auto &k: res){
printf("%lld, ", k);
}
else if(cmd == kMsgCmdGetConfig) {
printf("\n");
}
}
}
else if(cmd == consts::kMsgCmdGetConfig) {
// response config
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}
else if(cmd == kMsgCmdStatus) {
else if(cmd == consts::kMsgCmdStatus) {
// response status
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}
else if(cmd == kMsgCmdPtz) {
else if(cmd == consts::kMsgCmdPtz) {
// response Ptz control
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_response + rid, 0, "OK", js.contains(consts::kMsgCmd)?js[consts::kMsgCmd].as<string>():string(""), js.contains(consts::kMsgRid)?js[consts::kMsgRid].as<string>():string(""), data);
}else if(cmd == "list_videos"){
if(gRecFilesList != nullptr)
print_ts_files(*gRecFilesList);
else{
spdlog::warn("no local video files");
}
}
else {
auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, msg_field(js, kMsgRid), json());
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, msg_field(js, consts::kMsgRid), json());
}
}
catch(exception &e) {
auto str = string("exception: ") + e.what() + string(", ") + msg;
spdlog::error(str);
report_response_args(pub_topic, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
MqttMgr::report_response_args(gMqttClient, consts::pub_topic_report, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
}
//(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}
///
void start_mqtt(void *arg)
MqttHelper* start_mqtt(void *arg)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
auto client = MqttMgr::get_instance(consts::mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
spdlog::error("failed to create mqtt instance to {}", consts::mqtt_url);
return nullptr;
}
client->subscribe(sub_topic + dev_sn, handle_mqtt_req, false);
client->subscribe(consts::sub_topic + dev_sn, handle_mqtt_req, false);
return client;
}
//
......@@ -665,7 +657,8 @@ void readConfigFile(json &config){
void create_sd_directories(){
// create record paths
for(auto &k:strSDCardReservedPaths){
for(auto k:strSDCardReservedPaths){
spdlog::info("create path: {}", k);
system((string("mkdir -p ") + k).c_str());
}
}
......@@ -801,7 +794,7 @@ int main(int argc, char *argv[])
exit(1);
}
/// TODO: configuration issues handling
if(msg_field(gJsonConfig, kMsgConfigVgw).empty()){
if(msg_field(gJsonConfig, consts::kMsgConfigVgw).empty()){
spdlog::error("missing vgw config");
}
......@@ -818,7 +811,7 @@ int main(int argc, char *argv[])
start_md_bd(&args);
/// subscribe to mqtt
start_mqtt(nullptr);
gMqttClient = start_mqtt(nullptr);
thread thSmart = thread(maq_smart_task_entry, &args);
thSmart.join();
......
hi3518/main.mqtt.cc 0 → 100644
浏览文件 @ b92bb722
#include <iostream>
#include <string>
#include <chrono>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include "ref-memory.hpp"
#include <maque_system.h>
#include <maque_video_enc.h>
#include <maque_time.h>
#include <evpacket.h>
#include "common.h"
#include "ntp.h"
#include <time.h>
#include "raw_tcp.h"
#include "smart.h"
#include "motion.h"
#include <thread>
#include "mqttmgr.hpp"
#include "utils.hpp"
#include <zmq.h>
using namespace std;
using namespace jsoncons;
using namespace evutils;
///
#define NUM_IFRAME_PICK 2
#define NUM_MAX_QUEQUE_SIZE 60
#define NUM_MAX_PACKET_BYTES 1200
#define INTERVAL_LOG_QFULL 3 // its about 60/15 *3 secs
/// device info
static char dev_sn[TERMINAL_SN_SIZE] = {0};
/// default video gateway info
static int videoQuality = MAQUE_IMG_QUALITY_BETTER;
static uint64_t frameCntTotal = 0;
static uint32_t frameCntIframe = 0;
static uint32_t frameCntPframe = 0;
static uint64_t packetId = 0;
static spdlog::level::level_enum logLevel = spdlog::level::info;
static bool bPFrameAvail = false;
static bool bNeedRetry = false;
static int readFaildCnt = 0;
static bool enablePush = false;
static bool enableRecord = false;
static bool bCanRecord = false;
static bool bConnected = false;
static queue<DataItem> _frameQueue;
static mutex _mutFrame;
static condition_variable _condFrame;
static Notifier _notiFrame = {&_mutFrame, &_condFrame};
static CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
static int raw_socket_ = 0;
static bool bGotTime = false;
// zmq
static void *pPubCtx = nullptr;
static void *pPub = nullptr;
static const string strPubUrl = "inproc://frame";
/// flags used to stop modules
/// 0: stopped, 1: requested to stop, 2: runningstatic int
typedef struct ev_region_t {
float minx;
float miny;
float maxx;
float maxy;
} ev_region_t;
typedef struct ev_module_config_t {
int stat_current;
ev_region_t region;
bool enabled;
union {
struct {
int interval; // in seconds
}record;
struct {
float face_thresh; //[0,1]
float human_thresh;
} ai;
struct motion{
int level; // 1 - 6
} motion;
} module;
} ev_module_config_t;
static ev_module_config_t gConfigRecord = {0};
static ev_module_config_t gConfigAI = {0};
static ev_module_config_t gConfigMotion = {0};
enum EV_MSG_ERROR_CODE {
EV_MSG_ERROR_NONE,
EV_MSG_ERROR_EXCEPTION,
EV_MSG_ERROR_UNSUPPORTED_CMD,
EV_MSG_ERROR_CONTENT_SYNTAX,
EV_MSG_ERROR_NUM
};
void clean_up(int sig)
{
if(sig == SIGPIPE) {
// vgw connection error
bConnected = false;
bNeedRetry = true;
return;
}
spdlog::warn("clearn up called, with sig: {}", sig);
if (args.recFD) {
fclose(args.recFD);
}
if (raw_socket_) {
::close(raw_socket_);
}
// while (args.dataq->size() > 0) {
// DataItem elem = args.dataq->front();
// args.dataq->pop();
// MaQueVideoEncFrameInfo_s *pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
// MaQue_Demo_Mem_release(pMem->handleMem);
// }
//stop_md_bd();
LibXmMaQue_System_destroy();
exit(1);
}
bool is_big_endian(void)
{
union {
uint32_t i;
char c[4];
} bint = {0x01020304};
return bint.c[0] == 1;
}
MaQueVideoEncodeCfg_s *init_stream_cfg(MaQueVideoEncodeCfg_s *cfg)
{
cfg->eVidComp = (MaQueVideoCompress_e)-1;
cfg->eCapSize = (MaQueCaptureSize_e)-1;
cfg->eBitrateCtrl = (MaQueBitrateCtrl_e)-1;
cfg->eImageQuality = (MaQueImageQuality_e)-1;
cfg->nFps = -1;
cfg->nBitRate = -1;
cfg->nIFrameInterval = -1;
cfg->iQpMin = -1;
cfg->iQpMax = -1;
return cfg;
}
int32_t configure_stream(MaQueStreamChannel_e eStreamChn, MaQueVideoEncodeCfg_s *cfg)
{
int32_t res;
MaQueVideoEncodeCfg_s oldCfg;
if (LibXmMaQue_VideoEnc_getCfg(0, eStreamChn, &oldCfg)) {
spdlog::error("LibXmMaQue_VideoEncode_getCfg error");
return -1;
}
if (cfg->eVidComp != -1) {
oldCfg.eVidComp = cfg->eVidComp;
}
if (cfg->eCapSize != -1) {
oldCfg.eCapSize = cfg->eCapSize;
}
if (cfg->eBitrateCtrl != -1) {
oldCfg.eBitrateCtrl = cfg->eBitrateCtrl;
}
if (cfg->eImageQuality != -1) {
oldCfg.eImageQuality = cfg->eImageQuality;
}
if (cfg->nFps != -1) {
oldCfg.nFps = cfg->nFps;
}
if (cfg->nBitRate != -1) {
oldCfg.nBitRate = cfg->nBitRate;
}
if (cfg->nIFrameInterval != -1) {
oldCfg.nIFrameInterval = cfg->nIFrameInterval;
}
if (cfg->iQpMin > 0 && cfg->iQpMin <= 51 && cfg->iQpMax > 0 && cfg->iQpMax <= 51) {
oldCfg.iQpMin = cfg->iQpMin;
oldCfg.iQpMax = cfg->iQpMax;
oldCfg.bUseUserQp = 1;
}
spdlog::info("main: fps={}", oldCfg.nFps);
res = LibXmMaQue_VideoEnc_setCfg(0, eStreamChn, &oldCfg);
if (!res) {
memcpy(cfg, &oldCfg, sizeof(oldCfg));
}
return res;
}
//
void frame_free(void *data, void*hint)
{
MaQueVideoEncFrameInfo_s *frame = (MaQueVideoEncFrameInfo_s *)hint;
MaQue_Demo_Mem_release(frame->handleMem);
}
//
XM_S32 cb_frame_proc(XM_VOID *pUserArg, MaQueVideoEncFrameInfo_s *frame)
{
MaQueSystemTime_s *pstSysTime = &frame->stTimeStamp.stSysTime;
CallBackArg *args = (CallBackArg *)pUserArg;
struct timeval stTimeVal;
int ret;
frameCntTotal++;
zmq_msg_t msg;
void *hint = NULL;
if(frame->eEncodeType == MAQUE_ENCODE_TYPE_H264 && frame->eFrameType == MAQUE_FRAME_TYPE_VIDEO && frame->nDataLen > 0) {
evpacket_t* pkt = (evpacket_t*)((char*)frame->pData - sizeof(evpacket_t));
pkt->meta.magic[0] = 0xBE;
pkt->meta.magic[1] = 0xEF;
memcpy(pkt->meta.sn, dev_sn, sizeof(dev_sn));
/// NOTES: since both ends use little-endian, we don't use htonl, htons to save clocks
pkt->vpara.frame_type = frame->eSubType + 1;
pkt->vpara.packet_type = frame->eEncodeType;
pkt->vpara.res.width = frame->nWidth;
pkt->vpara.res.height = frame->nHeight;
/// TODO:
pkt->vpara.fps = 15;
//pkt->vpara.ts = frame->stTimeStamp;
pkt->length = frame->nDataLen;
pkt->meta.packet_id = packetId;
pkt->meta.crc = 0;
pkt->meta.crc = crc16((unsigned char*)pkt, sizeof(evpacket_t));
if(frame->eSubType == MAQUE_FRAME_SUBTYPE_I) {
frameCntIframe++;
}
else if(frame->eSubType == MAQUE_FRAME_SUBTYPE_P) {
frameCntPframe++;
}
zmq_msg_init_data(&msg, (char*)pkt, frame->nDataLen + sizeof(evpacket_t), nullptr, nullptr);
zmq_msg_send(&msg, pPub, 0);
packetId++;
}
MaQue_Demo_Mem_release(frame->handleMem);
return XM_SUCCESS;
}
void frame_send_entry(void *args)
{
int ret = 0;
// setup sub
void *pSub = zmq_socket(pPubCtx, ZMQ_SUB);
if(pSub == nullptr) {
spdlog::error("failed to create frame sending socket");
exit(1);
}
// ret = 1;
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
// ret = 2;
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_connect(pSub, strPubUrl.c_str());
if(ret != 0) {
spdlog::error("failed connect frame pub: {}", strPubUrl);
exit(1);
}
ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
spdlog::info("zmq sub successed (for frame sending)");
zmq_msg_t msg;
while (1) {
if(!bConnected||raw_socket_ <= 0) {
if(frameCntTotal % (NUM_MAX_QUEQUE_SIZE * 3) == 0) {
bNeedRetry = true;
if(raw_socket_) {
::close(raw_socket_);
raw_socket_ = 0;
}
}
if(bNeedRetry == true) {
// try reconect
bNeedRetry = false;
spdlog::info("try reconnecting to video gateway {}:{}", host, port);
auto ret = raw_connect(host, port, &raw_socket_);
if (ret < 0 || raw_socket_ <= 0) {
spdlog::error("failed to reconnect {}:{}", host, port);
bConnected = false;
}
else {
spdlog::info("successfully connected to {}:{}", host, port);
bConnected = true;
//LibXmMaQue_OSD_showTime(0, MAQUE_STREAM_CHN_MAIN, &stOsdTimeModParamMain.stOsdParam);
//freshConnect = pvArg->dataq->size();
/// pop all frames
}
}
continue;
}
zmq_msg_init(&msg);
ret = zmq_msg_recv(&msg, pSub, 0);
if(ret < 0) {
spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
continue;
}
ssize_t sz = zmq_msg_size(&msg);
char *ptr = (char*)zmq_msg_data(&msg);
ssize_t sent = 0;
ssize_t totalSent = 0;
while (sz > 0) {
ptr += sent;
auto tosent = sz > NUM_MAX_PACKET_BYTES? NUM_MAX_PACKET_BYTES:sz;
sent = ::send(raw_socket_, ptr, tosent, (tosent > sz? MSG_MORE:0)|MSG_NOSIGNAL);
if (sent <= 0) {
break;
}
sz -= sent;
totalSent += sent;
}
if (sent <= 0) {
spdlog::error("failed to send, peer down: {}", sz);
/// TODO: reconnection
bConnected = false;
bNeedRetry = true;
}
zmq_msg_close(&msg);
}
zmq_close (pSub);
}
using namespace chrono_literals;
void record_video_entry(void *args)
{
int ret = 0;
// this thread will never exit
while(1) {
if(!is_sdcard_avail()) {
spdlog::error("no sd card valial. can't record local videos");
this_thread::sleep_for(10s);
continue;
}
// setup sub
void *pSub = zmq_socket(pPubCtx, ZMQ_SUB);
// ret = 1;
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE, &ret, sizeof (ret));
// ret = 2;
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_IDLE, &ret, sizeof (ret));
// zmq_setsockopt(pSub, ZMQ_TCP_KEEPALIVE_INTVL, &ret, sizeof (ret));
ret = zmq_connect(pSub, strPubUrl.c_str());
if(ret != 0) {
spdlog::error("failed connect frame pub: {}", strPubUrl);
exit(1);
}
ret = zmq_setsockopt(pSub, ZMQ_SUBSCRIBE, "", 0);
spdlog::info("zmq sub successed (record)");
zmq_msg_t msg;
long long preTs = 0;
long long nowTs = 0;
uint32_t cnt = 0;
fstream * fp = nullptr;
while (1) {
zmq_msg_init(&msg);
ret = zmq_msg_recv(&msg, pSub, 0);
if(ret < 0) {
spdlog::error("failed to recv zmq msg: {}", zmq_strerror(ret));
continue;
}
ssize_t sz = zmq_msg_size(&msg);
char *ptr = (char*)zmq_msg_data(&msg);
/// we calc timestamps about every 2s
if(cnt % 30 == 0) {
nowTs = chrono::system_clock::now().time_since_epoch().count()/1000;
}
if(nowTs - preTs > gConfigRecord.module.record.interval*1000) {
if(fp != nullptr) {
if(fp->is_open())
fp->close();
delete fp;
}
fp = new fstream(recFilePath+to_string(nowTs), std::ios::out|std::ios::binary);
if(!fp->is_open()){
spdlog::error("failed to open record file");
delete fp;
fp = nullptr;
}
preTs = nowTs;
}
if(fp != nullptr) {
fp->write(ptr, sz);
}
cnt++;
zmq_msg_close(&msg);
}
zmq_close (pSub);
/// DO NOT! do this
//zmq_ctx_destroy(pPubCtx);
}
}
/// verify request message
string verify_request(json &js)
{
string ret;
string check[] = {kMsgCmd,kMsgData, kMsgRid, kMsgTime, kMsgRid};
for(auto &k: check) {
if(!js.contains(k)) {
ret += k + ",";
}
}
if(!ret.empty()) {
ret = "missing fields in request: " + ret;
}
return ret;
}
/// low level api for report and response
void _report_response(string topic, string message)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
}
auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
if(rc < 0) {
spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
}
else {
spdlog::info("successfully pub message at {}:{}", topic, message);
}
}
/// having data field along with code and msg
/// signature {"rid", "code", "msg", "cata", "data":{}}
void report_response_json(string topic, json& message)
{
// time
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
message.insert_or_assign(kMsgTime, ts);
// rid
auto rid = message["rid"].as<string>();
message.insert_or_assign(kMsgRid, rid);
// type
if(rid.empty()) {
message.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
message.insert_or_assign(kMsgType, kMsgTypeResponse);
}
// sn
message.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, message.to_string());
}
/// no data field. only code and msg field
void report_response_args(string topic, int code, string message, string cata, string rid, json &&data)
{
json js;
/// current timestamp
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
js.insert_or_assign(kMsgTime, ts);
/// rid
js.insert_or_assign(kMsgRid, rid);
/// type
if(rid.empty()) {
js.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
js.insert_or_assign(kMsgType, kMsgTypeResponse);
}
js.insert_or_assign(kMsgCata, cata);
///
js.insert_or_assign(kMsgCode, code);
js.insert_or_assign(kMsgMsg, message);
js.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, js.to_string());
}
int apply_config(json &js)
{
auto hasSD = is_sdcard_avail();
/// verify parameters
string str, strWarn;
if(!js.contains(kMsgData)) {
str = fmt::format("no data object in config: {}", js.to_string());
}
else {
json data = js[kMsgData];
string veri[] = {kMsgConfigVgw, kMsgConfigMqtt, kMsgConfigUpload};
for(auto &k:veri) {
auto value = msg_field(data, k);
if(value.empty()) {
str = fmt::format("no {} field in data: {}", k, js.to_string());
break;
}
else {
// parse host and port
auto uri = httplib::Uri::Parse(value);
if(uri.Host.empty()|| uri.Port.empty()) {
str = fmt::format("host addr invalid {}:{}, {}", k, value, js.to_string());
break;
}
// check network connections
int socket = 0;
auto rc = raw_connect(uri.Host, uri.Port, &socket);
if(rc < 0 || socket == 0) {
str = fmt::format("failed to connect to {}:{}, {}", k, value, js.to_string());
break;
}
else {
// we must close the test socket
::close(socket);
}
}
}
// check features field
json features = data.get_value_or<json>(kMsgConfigFeatures, json());
string feats[]= {kMsgConfigMotion, kMsgConfigRecord, kMsgConfigAi};
for(auto &k:feats) {
auto f = features.get_value_or<json>(k, json());
if(f.size() == 0) {
strWarn = fmt::format("{}, ", k);
}
if(!strWarn.empty()) {
strWarn = "no below feature in config, default configuration is applied: " + strWarn;
}
}
}
if(!str.empty()) {
report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, kMsgCmdConfig, msg_field(js, kMsgRid), json());
spdlog::error(str);
return -EV_MSG_ERROR_CONTENT_SYNTAX;
}
// try to apply
}
/// handles all incoming request from cloud services
void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
{
char msg[len];
memcpy(msg, data, len);
try {
auto js = json::parse(msg);
spdlog::info("received request on {}:\n{}", topic, (char*)msg);
auto str = verify_request(js);
if(!str.empty()) {
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(kMsgCmd)?js[kMsgCmd].as<string>():string(""), js.contains(kMsgRid)?js[kMsgRid].as<string>():string(""), json());
return;
}
/// dispatch msg
auto cmd = js[kMsgCmd].as<string>();
if(cmd == kMsgCmdConfig) {
// apply config
}
else if(cmd == kMsgCmdUploadVideo) {
// upload video
}
else if(cmd == kMsgCmdGetConfig) {
// response config
}
else if(cmd == kMsgCmdStatus) {
// response status
}
else if(cmd == kMsgCmdPtz) {
// response Ptz control
}
else {
auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, msg_field(js, kMsgRid), json());
}
}
catch(exception &e) {
auto str = string("exception: ") + e.what() + string(", ") + msg;
spdlog::error(str);
report_response_args(pub_topic, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
}
//(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}
///
void start_mqtt(void *arg)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
}
client->subscribe(sub_topic + dev_sn, handle_mqtt_req, false);
}
///
int main(int argc, char *argv[])
{
int ret = XM_SUCCESS;
spdlog::info("test on hi3518ev300, {}", is_big_endian());
get_mac_addr(dev_sn);
// get env for host and port
auto host_ = getenv("VGW_HOST");
enablePush = true;
if(host_) {
host = host_;
}
else {
enablePush = false;
}
auto port_ = getenv("VGW_PORT");
if(port_) {
port_ = port;
}
else {
enablePush = false;
}
/// TODO: check sd availability
bool bSDAvail = is_sdcard_avail();
auto logLevel_ = getenv("LOG_LEVEL");
if(logLevel_) {
logLevel = spdlog::level::from_str(logLevel_);
spdlog::set_level(logLevel);
}
else {
spdlog::info("log level configured to info");
spdlog::set_level(logLevel);
}
auto videoQuality_ = getenv("QUALITY");
if(videoQuality_) {
videoQuality = atoi(videoQuality_);
if(videoQuality >= MAQUE_IMG_QUALITY_NR || videoQuality < 0) {
videoQuality = MAQUE_IMG_QUALITY_BETTER;
}
}
MaQue_Demo_Mem_Init();
/// thread of utp time
auto thNtp = thread([] {
bool bGotTime = false;
time_t stm;
while(!bGotTime)
{
if(getNtpTime(&stm) >=0) {
spdlog::info("ntp got time");
::stime(&stm);
bGotTime = true;
}
else {
spdlog::warn("failed to get ntp time");
}
this_thread::sleep_for(chrono::seconds(20));
}
});
if(thNtp.joinable()) {
thNtp.detach();
}
MaQueStartParam_s startParam = {MAQUE_VIDEO_STANDARD_PAL, {MAQUE_VIDEO_COMPRESS_H264, MAQUE_VIDEO_COMPRESS_H265}, "/mnt/sd/Config/"};
ret = LibXmMaQue_System_startUp(&startParam);
spdlog::info("ret: {}", ret);
//
MaQueMemoryApi_s memApi = {MaQue_Demo_Mem_alloc, MaQue_Demo_Mem_release, MaQue_Demo_Mem_addRef, MaQue_Demo_Mem_setLength};
ret = LibXmMaQue_Mem_init(&memApi);
spdlog::info("mem ret: {}", ret);
MaQueSystemTime_s tm;
LibXmMaQue_Time_getCurrentTime(&tm);
LibXmMaQue_Time_print(&tm, nullptr);
//
MaQueCodeAbilities_s capb;
ret = LibXmMaQue_VideoEnc_getAbilities(0, &capb);
spdlog::info("abt ret: {}", ret);
if (XM_SUCCESS == ret) {
spdlog::info(fmt::format("capbilities: supported codecs {0:#b},"
"max fhd/s {0:d}, max res: {0:d}, per ch mres: {0:d}, {0:d}, {0:d}, {0:d}",
int(capb.videoEncTypeMask),
int(capb.maxEncPowerX1080P), int(capb.eDecImageSizeMax), int(capb.astVidEncChnAbility[0].eCapSizeMax),
int(capb.astVidEncChnAbility[1].eCapSizeMax), int(capb.astVidEncChnAbility[2].eCapSizeMax), int(capb.astVidEncChnAbility[3].eCapSizeMax)));
}
//
MaQueVideoEncodeCfg_s cfg;
init_stream_cfg(&cfg);
cfg.nFps = 15;
cfg.eIFrmIntvType = IFRAME_INTV_TYPE_TIME;
cfg.nIFrameInterval = 10;
cfg.eImageQuality = (MaQueImageQuality_e)videoQuality;
cfg.eVidComp = MAQUE_VIDEO_COMPRESS_H264;
cfg.nBitRate = 200 * 8; // 200KB
ret = configure_stream((MaQueStreamChannel_e)0, &cfg);
spdlog::info("cfg stream ret: {}", ret);
///
/// record setting
ret = LibXmMaQue_VideoEnc_startStream(0, (MaQueStreamChannel_e)0, cb_frame_proc, (void *)&args);
if (XM_SUCCESS == ret) {
spdlog::info("created record task successfully");
}
signal(SIGINT, clean_up);
// signal(SIGTERM, clean_up);
signal(SIGKILL, clean_up);
signal(SIGPIPE, SIG_IGN);
// setup zmq
pPubCtx = zmq_ctx_new();
pPub = zmq_socket(pPubCtx, ZMQ_PUB);
ret = zmq_bind(pPub, strPubUrl.c_str());
if(ret < 0) {
spdlog::error("{} failed to create zmq pub topic: {}", strPubUrl);
exit(1);
}
//
if (enablePush=true) {
ret = raw_connect(host, port, &raw_socket_);
if (ret < 0 || raw_socket_ <= 0) {
spdlog::error("failed connect to video gateway at {}:{}", host, port);
bConnected = false;
}
else {
bConnected = true;
spdlog::info("successfully connected to video gateway at {}:{}", host, port);
}
thread thPush = thread(frame_send_entry, &args);
spdlog::info("sizeof pkt header {}, sizeof tv {}", sizeof(evpacket_t), sizeof(timeval));
thPush.detach();
}
else {
spdlog::warn("push video is not enabled in environ settings");
}
start_md_bd(&args);
/// subscribe to mqtt
start_mqtt(nullptr);
thread thSmart = thread(maq_smart_task_entry, &args);
thSmart.join();
}
\ No newline at end of file
hi3518/main.raw.cc 0 → 100644
浏览文件 @ b92bb722
#include <iostream>
#include <string>
#include <chrono>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include "ref-memory.hpp"
#include <maque_system.h>
#include <maque_video_enc.h>
#include <maque_time.h>
#include <evpacket.h>
#include "common.h"
#include "ntp.h"
#include <time.h>
#include "raw_tcp.h"
#include "smart.h"
#include "motion.h"
#include <thread>
#include "mqttmgr.hpp"
#include "utils.hpp"
#include <zmq.h>
using namespace std;
using namespace jsoncons;
using namespace evutils;
///
#define NUM_IFRAME_PICK 2
#define NUM_MAX_QUEQUE_SIZE 60
#define NUM_MAX_PACKET_BYTES 1200
#define INTERVAL_LOG_QFULL 3 // its about 60/15 *3 secs
/// device info
static char dev_sn[13] = {0};
/// default video gateway info
static int videoQuality = MAQUE_IMG_QUALITY_BETTER;
static uint32_t frameCntTotal = 0;
static uint32_t frameCntIframe = 0;
static uint32_t frameCntPframe = 0;
static uint64_t packetId = 0;
static spdlog::level::level_enum logLevel = spdlog::level::info;
static bool bPFrameAvail = false;
static bool bNeedRetry = false;
static int readFaildCnt = 0;
static bool enablePush = false;
static bool enableRecord = false;
static bool bCanRecord = false;
static bool bConnected = false;
static queue<DataItem> _frameQueue;
static mutex _mutFrame;
static condition_variable _condFrame;
static Notifier _notiFrame = {&_mutFrame, &_condFrame};
static CallBackArg args = {nullptr, &_notiFrame, &_frameQueue};
static int raw_socket_ = 0;
static bool bGotTime = false;
// zmq
static void *pPubCtx = nullptr;
static void *pPub = nullptr;
static const string strPubUrl = "inproc://frame";
enum EV_MSG_ERROR_CODE {
EV_MSG_ERROR_NONE,
EV_MSG_ERROR_EXCEPTION,
EV_MSG_ERROR_UNSUPPORTED_CMD,
EV_MSG_ERROR_CONTENT_SYNTAX,
EV_MSG_ERROR_NUM
};
void clean_up(int sig)
{
spdlog::warn("clearn up called, with sig: {}", sig);
if (args.recFD) {
fclose(args.recFD);
}
if (raw_socket_) {
::close(raw_socket_);
}
// while (args.dataq->size() > 0) {
// DataItem elem = args.dataq->front();
// args.dataq->pop();
// MaQueVideoEncFrameInfo_s *pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
// MaQue_Demo_Mem_release(pMem->handleMem);
// }
//stop_md_bd();
LibXmMaQue_System_destroy();
exit(1);
}
bool is_big_endian(void)
{
union {
uint32_t i;
char c[4];
} bint = {0x01020304};
return bint.c[0] == 1;
}
MaQueVideoEncodeCfg_s *init_stream_cfg(MaQueVideoEncodeCfg_s *cfg)
{
cfg->eVidComp = (MaQueVideoCompress_e)-1;
cfg->eCapSize = (MaQueCaptureSize_e)-1;
cfg->eBitrateCtrl = (MaQueBitrateCtrl_e)-1;
cfg->eImageQuality = (MaQueImageQuality_e)-1;
cfg->nFps = -1;
cfg->nBitRate = -1;
cfg->nIFrameInterval = -1;
cfg->iQpMin = -1;
cfg->iQpMax = -1;
return cfg;
}
int32_t configure_stream(MaQueStreamChannel_e eStreamChn, MaQueVideoEncodeCfg_s *cfg)
{
int32_t res;
MaQueVideoEncodeCfg_s oldCfg;
if (LibXmMaQue_VideoEnc_getCfg(0, eStreamChn, &oldCfg)) {
spdlog::error("LibXmMaQue_VideoEncode_getCfg error");
return -1;
}
if (cfg->eVidComp != -1) {
oldCfg.eVidComp = cfg->eVidComp;
}
if (cfg->eCapSize != -1) {
oldCfg.eCapSize = cfg->eCapSize;
}
if (cfg->eBitrateCtrl != -1) {
oldCfg.eBitrateCtrl = cfg->eBitrateCtrl;
}
if (cfg->eImageQuality != -1) {
oldCfg.eImageQuality = cfg->eImageQuality;
}
if (cfg->nFps != -1) {
oldCfg.nFps = cfg->nFps;
}
if (cfg->nBitRate != -1) {
oldCfg.nBitRate = cfg->nBitRate;
}
if (cfg->nIFrameInterval != -1) {
oldCfg.nIFrameInterval = cfg->nIFrameInterval;
}
if (cfg->iQpMin > 0 && cfg->iQpMin <= 51 && cfg->iQpMax > 0 && cfg->iQpMax <= 51) {
oldCfg.iQpMin = cfg->iQpMin;
oldCfg.iQpMax = cfg->iQpMax;
oldCfg.bUseUserQp = 1;
}
spdlog::info("main: fps={}", oldCfg.nFps);
res = LibXmMaQue_VideoEnc_setCfg(0, eStreamChn, &oldCfg);
if (!res) {
memcpy(cfg, &oldCfg, sizeof(oldCfg));
}
return res;
}
//
XM_S32 cb_frame_proc(XM_VOID *pUserArg, MaQueVideoEncFrameInfo_s *frame)
{
MaQueSystemTime_s *pstSysTime = &frame->stTimeStamp.stSysTime;
CallBackArg *args = (CallBackArg *)pUserArg;
struct timeval stTimeVal;
int ret;
frameCntTotal++;
if(logLevel <= spdlog::level::debug) {
printf("\n\nnew frame: %u, p0 h:%08X\n", frameCntTotal, (int)frame->handleMem);
}
if (bCanRecord) {
auto nWrite = fwrite(frame->pData, 1, frame->nDataLen, args->recFD);
if(frame->nDataLen != nWrite) {
spdlog::error("failed to write file");
if(args->recFD) {
fclose(args->recFD);
}
}
}
if(enablePush && bConnected == false && frameCntTotal % (NUM_MAX_QUEQUE_SIZE * 3) == 0) {
bNeedRetry = true;
bConnected = false;
if(raw_socket_) {
::close(raw_socket_);
raw_socket_ = 0;
}
}
//printf("\n==== prepare frame with enablePush:%d, dataQ: %08hhX, frametype: %d\n", enablePush, (int)args->dataq, frame->eEncodeType);
if (enablePush && args->dataq && frame->eEncodeType == MAQUE_ENCODE_TYPE_H264) {
unique_lock<mutex> lock(*args->noti->mut);
// check delayed packets for about 12s. sufficiently high
if (args->dataq->size() >= NUM_MAX_QUEQUE_SIZE) {
spdlog::warn("dataq full");
volatile MaQueVideoEncFrameInfo_s * p1 = (MaQueVideoEncFrameInfo_s *)args->dataq->front().ud;
if(logLevel == spdlog::level::debug) {
printf("%u, p1 h:%08X\n", frameCntTotal, (int)p1->handleMem);
}
args->dataq->pop();
if(p1 == frame) {
spdlog::error("BUG!!! new frame address same as what in Queue");
}
MaQue_Demo_Mem_release(p1->handleMem);
p1 = nullptr;
}
if (frame->pData && frame->nDataLen > 0) {
timeval tv;
//::gettimeofday(&tv,NULL);
DataItem dt = {(char *)frame->pData - sizeof(evpacket_t), frame->nDataLen + sizeof(evpacket_t), (void *)frame};
evpacket_t *pkt = (evpacket_t *)dt.buf;
memset(pkt, 0, sizeof(*pkt));
pkt->meta.magic[0] = 0xBE;
pkt->meta.magic[1] = 0xEF;
memcpy(pkt->meta.sn, dev_sn, sizeof(dev_sn));
/// NOTES: since both ends use little-endian, we don't use htonl, htons to save clocks
pkt->vpara.frame_type = frame->eSubType + 1;
pkt->vpara.packet_type = frame->eEncodeType;
pkt->vpara.res.width = frame->nWidth;
pkt->vpara.res.height = frame->nHeight;
/// TODO:
pkt->vpara.fps = 15;
//pkt->vpara.ts = frame->stTimeStamp;
pkt->length = frame->nDataLen;
pkt->meta.packet_id = packetId;
pkt->meta.crc = 0;
pkt->meta.crc = crc16((unsigned char*)pkt, sizeof(evpacket_t));
if (frame->eSubType == MAQUE_FRAME_SUBTYPE_I) {
frameCntIframe++;
}
else if(frame->eSubType == MAQUE_FRAME_SUBTYPE_P) {
frameCntPframe++;
}
MaQue_Demo_Mem_addRef(frame->handleMem);
args->dataq->push(dt);
bPFrameAvail = true;
spdlog::debug("frame meter ic:{}, pc:{}, tc:{}, len:{}, cid:{}", frameCntIframe, frameCntPframe,frameCntTotal, frame->nDataLen, packetId);
packetId++;
}
else {
spdlog::error("invalid frame");
}
lock.unlock();
args->noti->cond->notify_one();
}
else {
//spdlog::warn("pusher is disabled {}:{}", __FILE__, __LINE__);
}
if(logLevel <= spdlog::level::debug) {
printf("%u, p2 h:%08X\n", frameCntTotal, (int)frame->handleMem);
}
MaQue_Demo_Mem_release(frame->handleMem);
return XM_SUCCESS;
}
void frame_send_entry(void *args)
{
CallBackArg *pvArg = (CallBackArg *)args;
int freshConnect = 0;
DataItem elem = {0};
while (1) {
MaQueVideoEncFrameInfo_s *pMem;
{
unique_lock<mutex> lk(*pvArg->noti->mut);
pvArg->noti->cond->wait(lk, [pvArg] { return !pvArg->dataq->empty();});
if (pvArg->dataq->empty()) {
spdlog::warn("dataq empty");
continue;
}
// /// remove old frames
// lk.unlock();
// while(pvArg->dataq->size() > 2) {
// lk.lock();
// elem = pvArg->dataq->front();
// pvArg->dataq->pop();
// lk.unlock();
// pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
// MaQue_Demo_Mem_release(pMem->handleMem);
// }
// lk.lock();
elem = pvArg->dataq->front();
pvArg->dataq->pop();
lk.unlock();
pMem = (MaQueVideoEncFrameInfo_s *)elem.ud;
}
if(bConnected == false) {
if(bNeedRetry == true) {
// try reconect
bNeedRetry = false;
if(raw_socket_ != 0) {
::close(raw_socket_);
raw_socket_ = 0;
}
spdlog::info("try reconnecting to video gateway {}:{}", host, port);
auto ret = raw_connect(host, port, &raw_socket_);
if (ret < 0 || raw_socket_ <= 0) {
spdlog::error("failed to reconnect {}:{}", host, port);
bConnected = false;
}
else {
spdlog::info("successfully connected to {}:{}", host, port);
bConnected = true;
//LibXmMaQue_OSD_showTime(0, MAQUE_STREAM_CHN_MAIN, &stOsdTimeModParamMain.stOsdParam);
//freshConnect = pvArg->dataq->size();
/// pop all frames
}
}
}
if (bConnected == false || !pMem||elem.size <= 0 || elem.buf == nullptr || (elem.buf + sizeof(evpacket_t)) == nullptr) {
//printf("invalid frame. addr: %08X, len:%u\n", (uint32_t)elem.buf, elem.size);
string msg;
if(elem.size <=0) {
msg = fmt::format("invalid frame size: {}", elem.size);
spdlog::warn(msg);
}
else if(elem.buf == nullptr||elem.buf + sizeof(evpacket_t) == nullptr) {
msg = fmt::format("invalid buffer addr: {0:p}", elem.buf);
spdlog::warn(msg);
}
else {
/// bConnected is false. needn't to log again
//msg = fmt::format("no connection to video gateway server {}:{}", host, port);
//spdlog::warn(msg);
}
/// release msg
if(pMem && pMem->handleMem) {
if(logLevel <= spdlog::level::debug) {
printf("%u, p3 h:%08X\n", frameCntTotal, (int)pMem->handleMem);
}
MaQue_Demo_Mem_release(pMem->handleMem);
pMem = nullptr;
}
continue;
}
/// send frame
if (elem.size > 0) {
auto sz = elem.size;
char *ptr = elem.buf;
ssize_t sent = 0;
while (elem.size > 0) {
ptr += sent;
sent = ::send(raw_socket_, ptr, elem.size > NUM_MAX_PACKET_BYTES? NUM_MAX_PACKET_BYTES:elem.size, 0);
if (sent <= 0) {
break;
}
else {
//spdlog::debug("sent:{}, total:{}", sent, sz);
}
elem.size -= sent;
}
if (sent <= 0) {
spdlog::error("failed to send, peer down");
/// TODO: reconnection
bConnected = false;
bNeedRetry = true;
}
}
else {
spdlog::error("frame size error");
}
if(logLevel <= spdlog::level::debug) {
printf("%u, p4 h:%08X\n", frameCntTotal, (int)pMem->handleMem);
}
MaQue_Demo_Mem_release(pMem->handleMem);
}
}
/// verify request message
string verify_request(json &js)
{
string ret;
string check[] = {kMsgCmd,kMsgData, kMsgRid, kMsgTime, kMsgRid};
for(auto &k: check) {
if(!js.contains(k)) {
ret += k + ",";
}
}
if(!ret.empty()) {
ret = "missing fields in request: " + ret;
}
return ret;
}
/// low level api for report and response
void _report_response(string topic, string message)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
}
auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
if(rc < 0) {
spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
}
else {
spdlog::info("successfully pub message at {}:{}", topic, message);
}
}
/// having data field along with code and msg
/// signature {"rid", "code", "msg", "cata", "data":{}}
void report_response_json(string topic, json& message)
{
// time
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
message.insert_or_assign(kMsgTime, ts);
// rid
auto rid = message["rid"].as<string>();
message.insert_or_assign(kMsgRid, rid);
// type
if(rid.empty()) {
message.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
message.insert_or_assign(kMsgType, kMsgTypeResponse);
}
// sn
message.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, message.to_string());
}
/// no data field. only code and msg field
void report_response_args(string topic, int code, string message, string cata, string rid, json &&data)
{
json js;
/// current timestamp
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
js.insert_or_assign(kMsgTime, ts);
/// rid
js.insert_or_assign(kMsgRid, rid);
/// type
if(rid.empty()) {
js.insert_or_assign(kMsgType, kMsgTypeReport);
}
else {
js.insert_or_assign(kMsgType, kMsgTypeResponse);
}
js.insert_or_assign(kMsgCata, cata);
///
js.insert_or_assign(kMsgCode, code);
js.insert_or_assign(kMsgMsg, message);
js.insert_or_assign(kMsgSn, string(dev_sn));
_report_response(topic, js.to_string());
}
int apply_config(json &js)
{
auto hasSD = is_sdcard_avail();
/// verify parameters
string str, strWarn;
if(!js.contains(kMsgData)) {
str = fmt::format("no data object in config: {}", js.to_string());
}
else {
json data = js[kMsgData];
string veri[] = {kMsgConfigVgw, kMsgConfigMqtt, kMsgConfigUpload};
for(auto &k:veri) {
auto value = msg_field(data, k);
if(value.empty()) {
str = fmt::format("no {} field in data: {}", k, js.to_string());
break;
}
else {
// parse host and port
auto uri = httplib::Uri::Parse(value);
if(uri.Host.empty()|| uri.Port.empty()) {
str = fmt::format("host addr invalid {}:{}, {}", k, value, js.to_string());
break;
}
// check network connections
int socket = 0;
auto rc = raw_connect(uri.Host, uri.Port, &socket);
if(rc < 0 || socket == 0) {
str = fmt::format("failed to connect to {}:{}, {}", k, value, js.to_string());
break;
}
else {
// we must close the test socket
::close(socket);
}
}
}
// check features field
json features = data.get_value_or<json>(kMsgConfigFeatures, json());
string feats[]= {kMsgConfigMotion, kMsgConfigRecord, kMsgConfigAi};
for(auto &k:feats) {
auto f = features.get_value_or<json>(k, json());
if(f.size() == 0) {
strWarn = fmt::format("{}, ", k);
}
if(!strWarn.empty()) {
strWarn = "no below feature in config, default configuration is applied: " + strWarn;
}
}
}
if(!str.empty()) {
report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, kMsgCmdConfig, msg_field(js, kMsgRid), json());
spdlog::error(str);
return -EV_MSG_ERROR_CONTENT_SYNTAX;
}
// try to apply
}
/// handles all incoming request from cloud services
void handle_mqtt_req(MqttHelper *hlp, const void * const data, int len, string topic)
{
char msg[len];
memcpy(msg, data, len);
try {
auto js = json::parse(msg);
spdlog::info("received request on {}:\n{}", topic, (char*)msg);
auto str = verify_request(js);
if(!str.empty()) {
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_CONTENT_SYNTAX, str, js.contains(kMsgCmd)?js[kMsgCmd].as<string>():string(""), js.contains(kMsgRid)?js[kMsgRid].as<string>():string(""), json());
return;
}
/// dispatch msg
auto cmd = js[kMsgCmd].as<string>();
if(cmd == kMsgCmdConfig) {
// apply config
}
else if(cmd == kMsgCmdUploadVideo) {
// upload video
}
else if(cmd == kMsgCmdGetConfig) {
// response config
}
else if(cmd == kMsgCmdStatus) {
// response status
}
else if(cmd == kMsgCmdPtz) {
// response Ptz control
}
else {
auto str = fmt::format("unsupported cmd {}: {}", cmd, js.to_string());
spdlog::warn(str);
report_response_args(pub_topic, EV_MSG_ERROR_UNSUPPORTED_CMD, str, cmd, msg_field(js, kMsgRid), json());
}
}
catch(exception &e) {
auto str = string("exception: ") + e.what() + string(", ") + msg;
spdlog::error(str);
report_response_args(pub_topic, EV_MSG_ERROR_EXCEPTION, str, string(""), string(""), json());
}
//(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}
///
void start_mqtt(void *arg)
{
auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("failed to create mqtt instance to {}", mqtt_url);
return;
}
client->subscribe(sub_topic + dev_sn, handle_mqtt_req, false);
}
///
int main(int argc, char *argv[])
{
int ret = XM_SUCCESS;
spdlog::info("test on hi3518ev300, {}", is_big_endian());
get_mac_addr(dev_sn);
// get env for host and port
auto host_ = getenv("VGW_HOST");
enablePush = true;
if(host_) {
host = host_;
}
else {
enablePush = false;
}
auto port_ = getenv("VGW_PORT");
if(port_) {
port_ = port;
}
else {
enablePush = false;
}
/// TODO: check sd availability
bool bSDAvail = is_sdcard_avail();
auto logLevel_ = getenv("LOG_LEVEL");
if(logLevel_) {
logLevel = spdlog::level::from_str(logLevel_);
spdlog::set_level(logLevel);
}
else {
spdlog::info("log level configured to info");
spdlog::set_level(logLevel);
}
auto videoQuality_ = getenv("QUALITY");
if(videoQuality_) {
videoQuality = atoi(videoQuality_);
if(videoQuality >= MAQUE_IMG_QUALITY_NR || videoQuality < 0) {
videoQuality = MAQUE_IMG_QUALITY_BETTER;
}
}
MaQue_Demo_Mem_Init();
/// thread of utp time
auto thNtp = thread([] {
bool bGotTime = false;
time_t stm;
while(!bGotTime)
{
if(getNtpTime(&stm) >=0) {
spdlog::info("ntp got time");
::stime(&stm);
bGotTime = true;
}
else {
spdlog::warn("failed to get ntp time");
}
this_thread::sleep_for(chrono::seconds(20));
}
});
if(thNtp.joinable()) {
thNtp.detach();
}
MaQueStartParam_s startParam = {MAQUE_VIDEO_STANDARD_PAL, {MAQUE_VIDEO_COMPRESS_H264, MAQUE_VIDEO_COMPRESS_H265}, "/mnt/sd/Config/"};
ret = LibXmMaQue_System_startUp(&startParam);
spdlog::info("ret: {}", ret);
//
MaQueMemoryApi_s memApi = {MaQue_Demo_Mem_alloc, MaQue_Demo_Mem_release, MaQue_Demo_Mem_addRef, MaQue_Demo_Mem_setLength};
ret = LibXmMaQue_Mem_init(&memApi);
spdlog::info("mem ret: {}", ret);
MaQueSystemTime_s tm;
LibXmMaQue_Time_getCurrentTime(&tm);
LibXmMaQue_Time_print(&tm, nullptr);
//
MaQueCodeAbilities_s capb;
ret = LibXmMaQue_VideoEnc_getAbilities(0, &capb);
spdlog::info("abt ret: {}", ret);
if (XM_SUCCESS == ret) {
spdlog::info(fmt::format("capbilities: supported codecs {0:#b},"
"max fhd/s {0:d}, max res: {0:d}, per ch mres: {0:d}, {0:d}, {0:d}, {0:d}",
int(capb.videoEncTypeMask),
int(capb.maxEncPowerX1080P), int(capb.eDecImageSizeMax), int(capb.astVidEncChnAbility[0].eCapSizeMax),
int(capb.astVidEncChnAbility[1].eCapSizeMax), int(capb.astVidEncChnAbility[2].eCapSizeMax), int(capb.astVidEncChnAbility[3].eCapSizeMax)));
}
//
MaQueVideoEncodeCfg_s cfg;
init_stream_cfg(&cfg);
cfg.nFps = 15;
cfg.eIFrmIntvType = IFRAME_INTV_TYPE_TIME;
cfg.nIFrameInterval = 10;
cfg.eImageQuality = (MaQueImageQuality_e)videoQuality;
cfg.eVidComp = MAQUE_VIDEO_COMPRESS_H264;
cfg.nBitRate = 200 * 8; // 200KB
ret = configure_stream((MaQueStreamChannel_e)0, &cfg);
spdlog::info("cfg stream ret: {}", ret);
///
/// record setting
if(enableRecord && bSDAvail) {
FILE *recFD = fopen(recFilePath, "w+");
if (!recFD) {
spdlog::error("fopen() Failed: {}", recFilePath);
bCanRecord = false;
}
else {
args.recFD = recFD;
bCanRecord = true;
}
}
ret = LibXmMaQue_VideoEnc_startStream(0, (MaQueStreamChannel_e)0, cb_frame_proc, (void *)&args);
if (XM_SUCCESS == ret) {
spdlog::info("created record task successfully");
}
signal(SIGINT, clean_up);
// signal(SIGTERM, clean_up);
signal(SIGKILL, clean_up);
// setup zmq
pPubCtx = zmq_ctx_new();
pPub = zmq_socket(pPubCtx, ZMQ_PUB);
ret = zmq_bind(pPub, strPubUrl.c_str());
if(ret < 0) {
spdlog::error("{} failed to create zmq pub topic: {}", strPubUrl);
exit(1);
}
//
if (enablePush) {
ret = raw_connect(host, port, &raw_socket_);
if (ret < 0 || raw_socket_ <= 0) {
spdlog::error("failed connect to video gateway at {}:{}", host, port);
bConnected = false;
}
else {
bConnected = true;
spdlog::info("successfully connected to video gateway at {}:{}", host, port);
}
thread thPush = thread(frame_send_entry, &args);
spdlog::info("sizeof pkt header {}, sizeof tv {}", sizeof(evpacket_t), sizeof(timeval));
thPush.detach();
}
else {
spdlog::warn("push video is not enabled in environ settings");
}
start_md_bd(&args);
/// subscribe to mqtt
start_mqtt(nullptr);
thread thSmart = thread(maq_smart_task_entry, &args);
thSmart.join();
}
\ No newline at end of file
hi3518/mqttmgr.hpp deleted 100644 → 0
浏览文件 @ 2c0181b7
/**
* Mqtt main implementation file
* Bruce.Lu 20200506
* sub topic for request: evcamera/v1.0/request/<SN>
* pub topic for report/response: evcamera/v1.0/response
*/
#ifndef __EV_MQTT_MGR___
#define __EV_MQTT_MGR___
#include <mqtt_helper.hpp>
#include <mutex>
#include <map>
using namespace std;
/// just a simple wrapper around MqttHelper
class MqttMgr {
public:
static map<string, MqttHelper *> insts;
static mutex mut;
public:
static MqttHelper* get_instance(string mqtt_url, string devsn){
auto key = mqtt_url + "/" + devsn;
lock_guard<mutex> lg(MqttMgr::mut);
try{
if(MqttMgr::insts.count(key) == 0){
auto inst = new MqttHelper(mqtt_url, "evc." + devsn);
MqttMgr::insts[key] = inst;
}else{
//
}
}catch(exception &e) {
spdlog::error("failed to create mqtt adaptor: {}", e.what());
}
return MqttMgr::insts[key];
}
static void remove(string mqtt_url, string devsn){
auto key = mqtt_url + "/" + devsn;
lock_guard<mutex> lg(MqttMgr::mut);
if(MqttMgr::insts.count(key) == 0){
}else{
delete MqttMgr::insts[key];
MqttMgr::insts.erase(key);
}
}
MqttMgr(){}
~MqttMgr(){
for(auto &x:insts){
if(x.second){
delete x.second;
}
insts.erase(x.first);
}
}
};
map<string, MqttHelper *> MqttMgr::insts;
mutex MqttMgr::mut;
#endif
hi3518/ntp.cc
浏览文件 @ b92bb722
......@@ -84,6 +84,22 @@ int getNtpTime(time_t * txTm)
last_errno = errno;
continue;
}
struct timeval timeout;
timeout.tv_sec = 2;
timeout.tv_usec = 0;
if (setsockopt (socket_, SOL_SOCKET, SO_RCVTIMEO, (char *)&timeout, sizeof(timeout)) < 0) {
spdlog::error("setsockopt SO_RCVTIMEO failed");
return -1;
}
timeout.tv_sec = 2;
if (setsockopt (socket_, SOL_SOCKET, SO_SNDTIMEO, (char *)&timeout, sizeof(timeout)) < 0) {
spdlog::error("setsockopt SO_SNDTIMEO failed");
return -1;
}
rv = ::connect(socket_, rp->ai_addr, rp->ai_addrlen);
if (rv == 0) {
auto addr = (struct sockaddr_in *)rp->ai_addr;
......
hi3518/smart.cc
浏览文件 @ b92bb722
#include "smart.h"
#include <spdlog/spdlog.h>
extern "C"
{
......@@ -7,6 +8,8 @@ extern "C"
#include <maque_osd.h>
}
#include "mqtt_helper.hpp"
#define DIV_UP(x, a) ( ((x) + ((a) - 1) ) / a )
static int smartCnt = 0;
......
hi3518/smart.h
浏览文件 @ b92bb722
#ifndef __SMART_H__
#define __SMART_H__
#ifndef __EV_SMART_HPP__
#define __EV_SMART_HPP__
#pragma once
void maq_smart_task_entry(void *pArg);
#endif
\ No newline at end of file
include/mqtt_helper.cc 0 → 100644
浏览文件 @ b92bb722
/**
* mqtt_helper.hpp
*
* simplifies api for mqtt
*
* Bruce.Lu @20200415
*/
#include <mqtt_helper.hpp>
#include <string>
#include <chrono>
#include <thread>
#include <set>
#include <map>
#include <future>
#include <mutex>
#include <httplib.h>
// #define SPDLOG_ACTIVE_LEVEL SPDLOG_LEVEL_TRACE
// #define SPDLOG_DEBUG_ON
// #define SPDLOG_TRACE_ON
#include <spdlog/spdlog.h>
#include <fmt/format.h>
extern "C" {
#include <MQTTAsync.h>
}
using namespace std;
void on_connected(void* context, MQTTAsync_successData* response);
void on_connlost(void *context, char *cause);
int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message);
void on_disconn(void* context, MQTTAsync_successData* response);
void on_subscribed(void* context, MQTTAsync_successData* response);
void on_conn_fail(void* context, MQTTAsync_failureData* response);
void on_subscribed_fail(void* context, MQTTAsync_failureData* response);
void on_sent(void* context, MQTTAsync_successData* response);
void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response);
void on_unsubscribe_ok(void* context, MQTTAsync_successData* response);
class MqttHelper;
typedef void (*on_res_fun_ptr_t)(void *res);
typedef void (*on_msg_fun_ptr_t)(MqttHelper *hlp, const void * const data, int len, string topic);
template<typename R>
bool is_ready(std::future<R> const& f)
{
return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
}
void on_connlost(void *context, char *cause)
{
MqttHelper *self = (MqttHelper *) context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
spdlog::error("mqtt connection lost: {}", cause? cause: "unkown reason");
spdlog::warn("reconnecting {}", self->addr);
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
if ((rc = MQTTAsync_connect(self->client, &conn_opts)) != MQTTASYNC_SUCCESS) {
string msg = fmt::format("Failed to start connect:", MQTTAsync_strerror(rc));
spdlog::error(msg);
//self->state.set_exception(StrException(msg));
}
}
int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message)
{
int i;
char* payloadptr;
MqttHelper *self = (MqttHelper *) context;
(*self)(string(topic), message->payload, message->payloadlen);
MQTTAsync_freeMessage(&message);
MQTTAsync_free(topic);
return 1;
}
void on_disconn(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
// self->state = MqttHelper::State::Disconnected;
// if(self->state.get_future().valid()){
// self->state = std::promise<MqttHelper::State>();
// }
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::Disconnected;
self->state.set_value(as);
spdlog::info("successful disconnected {}", self->addr);
}
void on_subscribed(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
//spdlog::debug("subscribe succeeded");
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::SubscribeOK;
self->state.set_value(as);
}
void on_subscribed_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
//string msg = fmt::format("subscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
// spdlog::error(msg);
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::SubScribeFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_unsubscribe_ok(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::UnsubscribeOk;
self->state.set_value(as);
}
void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
string msg = fmt::format("unsubscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
spdlog::error(msg);
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::UnsubscribeFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_conn_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
string msg = fmt::format("Connect failed: {}", MQTTAsync_strerror(response ? response->code : 0));
spdlog::error(msg);
//self->state.set_exception(StrException(msg));
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::ConnFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_sent(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
spdlog::debug("Message delivery confirmed, tok: {}", response->token);
}
void on_connected(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::Ready;
self->state.set_value(as);
spdlog::info("connected to mqtt {}", self->addr);
}
map<string, MqttHelper *> MqttMgr::insts;
mutex MqttMgr::mut;
#ifdef TEST_MQTT
#include <httplib.h>
void my_on_msg1(MqttHelper *hlp, const void * const data, int len, string topic)
{
string msg;
msg.resize(len+1);
memcpy((void*)&msg[0], data, len);
*(char*)(&msg[0] +len) = 0;
spdlog::info("I1 recv on {}: {}", topic, msg);
(*hlp)[to_string(stoi(topic) + 1)]("hello ack1", 11);
}
void my_on_msg2(MqttHelper *hlp, const void * const data, int len, string topic)
{
string msg;
msg.resize(len+1);
memcpy((void*)&msg[0], data, len);
*(char*)(&msg[0] +len) = 0;
spdlog::info("I2 recv on {}: {}", topic, msg);
(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}
int main()
{
string mqtt_url = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
string mqtt_cid = "testaabbc";
auto tmpEnv = getenv("MQTT_URL");
if(tmpEnv) {
mqtt_url = string(tmpEnv);
}
tmpEnv = getenv("CID");
if(tmpEnv) {
mqtt_cid = string(tmpEnv);
}
spdlog::set_level(spdlog::level::debug);
{
try {
MqttHelper hlp(mqtt_url, mqtt_cid);
hlp["hello"].pub("hello", 6, 1);
hlp.subscribe("no_on_msg", NULL);
auto x = thread([&hlp] {
for(int i = 0; i < 20; i ++)
{
hlp.subscribe(to_string(i), my_on_msg1);
}
hlp["hello"].pub("hello1", 7, 1);
});
auto y = thread([&hlp] {
for(int i = 0; i < 10; i ++)
{
hlp.subscribe(to_string(i), my_on_msg2);
}
for(int i = 10; i < 30; i ++)
{
hlp.subscribe(to_string(i), my_on_msg2, true);
}
hlp["hello"].pub("hello2", 7, 1);
});
x.join();
y.join();
this_thread::sleep_for(chrono::seconds(20));
}
catch(exception &e) {
spdlog::error("failed to connect mqtt: {}, {}", mqtt_url, e.what());
exit(1);
}
}
this_thread::sleep_for(chrono::seconds(2));
return 0;
}
#endif
include/mqtt_helper.hpp
浏览文件 @ b92bb722
/**
* mqtt_helper.hpp
*
* simplifies api for mqtt
* simplified apis for mqtt
*
* Bruce.Lu @20200415
*/
#ifndef __MQTT_HELPER__
#define __MQTT_HELPER__
#pragma once
#include <string>
#include <chrono>
#include <thread>
#include <set>
#include <map>
#include <future>
#include <mutex>
#include <map>
#include <httplib.h>
// #define SPDLOG_ACTIVE_LEVEL SPDLOG_LEVEL_TRACE
// #define SPDLOG_DEBUG_ON
// #define SPDLOG_TRACE_ON
#include <spdlog/spdlog.h>
#include <fmt/format.h>
#include <spdlog/spdlog.h>
#include <jsoncons/json.hpp>
#include "evutils.hpp"
extern "C" {
#include <MQTTAsync.h>
}
using namespace std;
void on_connected(void* context, MQTTAsync_successData* response);
void on_connlost(void *context, char *cause);
int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message);
void on_disconn(void* context, MQTTAsync_successData* response);
void on_subscribed(void* context, MQTTAsync_successData* response);
void on_conn_fail(void* context, MQTTAsync_failureData* response);
void on_subscribed_fail(void* context, MQTTAsync_failureData* response);
void on_sent(void* context, MQTTAsync_successData* response);
void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response);
void on_unsubscribe_ok(void* context, MQTTAsync_successData* response);
using namespace jsoncons;
using namespace evutils;
extern void on_connected(void* context, MQTTAsync_successData* response);
extern void on_connlost(void *context, char *cause);
extern int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message);
extern void on_disconn(void* context, MQTTAsync_successData* response);
extern void on_subscribed(void* context, MQTTAsync_successData* response);
extern void on_conn_fail(void* context, MQTTAsync_failureData* response);
extern void on_subscribed_fail(void* context, MQTTAsync_failureData* response);
extern void on_sent(void* context, MQTTAsync_successData* response);
extern void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response);
extern void on_unsubscribe_ok(void* context, MQTTAsync_successData* response);
class MqttHelper;
typedef void (*on_res_fun_ptr_t)(void *res);
typedef void (*on_msg_fun_ptr_t)(MqttHelper *hlp, const void * const data, int len, string topic);
template<typename R>
bool is_ready(std::future<R> const& f)
{
return f.wait_for(std::chrono::seconds(0)) == std::future_status::ready;
}
bool is_ready(std::future<R> const& f);
struct StrException : public std::exception {
std::string s;
......@@ -106,6 +103,7 @@ private:
protected:
public:
string mqtt_url;
string id;
string addr;
MQTTAsync client;
enum class State {
......@@ -129,7 +127,7 @@ public:
/// State state = State::None; // 0: initial ; 1: ready; 2: disconnected; 3: destroyed
promise<AsyncResult> state;
MqttHelper(string mqtt_url, string id, int kai = 20, int cs = 1): mqtt_url(mqtt_url)
MqttHelper(string mqtt_url, string id, int kai = 20, int cs = 1): mqtt_url(mqtt_url), id(id)
{
// make connection, throw excpetions
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
......@@ -144,7 +142,7 @@ public:
spdlog::debug("mqtt url: proto: {}, host: {}, port: {}, user: {}, pass: {}", uri.Protocol, uri.Host, uri.Port, uri.User, uri.Password);
if(MQTTASYNC_SUCCESS != MQTTAsync_create(&client, addr.c_str(), id.c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL)) {
if(MQTTASYNC_SUCCESS != MQTTAsync_create(&client, addr.c_str(), (string("EVC")+id).c_str(), MQTTCLIENT_PERSISTENCE_NONE, NULL)) {
msg = "failed to async create mqtt";
spdlog::error(msg);
throw StrException(msg);
......@@ -270,7 +268,6 @@ public:
return 0;
}
~MqttHelper()
{
int rc;
......@@ -293,204 +290,141 @@ public:
};
void on_connlost(void *context, char *cause)
{
MqttHelper *self = (MqttHelper *) context;
MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
int rc;
spdlog::error("mqtt connection lost: {}", cause? cause: "unkown reason");
/// just a simple wrapper around MqttHelper
class MqttMgr {
public:
static map<string, MqttHelper *> insts;
static mutex mut;
public:
spdlog::warn("reconnecting {}", self->addr);
conn_opts.keepAliveInterval = 20;
conn_opts.cleansession = 1;
if ((rc = MQTTAsync_connect(self->client, &conn_opts)) != MQTTASYNC_SUCCESS) {
string msg = fmt::format("Failed to start connect:", MQTTAsync_strerror(rc));
spdlog::error(msg);
//self->state.set_exception(StrException(msg));
static MqttHelper* get_instance(string mqtt_url, string devsn){
auto key = mqtt_url + "/" + devsn;
lock_guard<mutex> lg(MqttMgr::mut);
try{
if(MqttMgr::insts.count(key) == 0){
auto inst = new MqttHelper(mqtt_url, "evc." + devsn);
MqttMgr::insts[key] = inst;
}else{
//
}
}catch(exception &e) {
spdlog::error("failed to create mqtt adaptor: {}", e.what());
}
}
int on_msgarrvd(void *context, char *topic, int topicLen, MQTTAsync_message *message)
{
int i;
char* payloadptr;
MqttHelper *self = (MqttHelper *) context;
string msg;
msg.resize(message->payloadlen + 1);
payloadptr = (char *)message->payload;
memcpy((void*)&msg[0], message->payload, message->payloadlen);
*(char*)(&msg[0] +message->payloadlen) = 0;
spdlog::debug("new message: {}, at {}", msg, topic);
(*self)(string(topic), message->payload, message->payloadlen);
MQTTAsync_freeMessage(&message);
MQTTAsync_free(topic);
return 1;
}
void on_disconn(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
// self->state = MqttHelper::State::Disconnected;
// if(self->state.get_future().valid()){
// self->state = std::promise<MqttHelper::State>();
// }
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::Disconnected;
self->state.set_value(as);
spdlog::info("successful disconnected {}", self->addr);
}
void on_subscribed(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
//spdlog::debug("subscribe succeeded");
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::SubscribeOK;
self->state.set_value(as);
}
void on_subscribed_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
//string msg = fmt::format("subscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
// spdlog::error(msg);
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::SubScribeFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_unsubscribe_ok(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::UnsubscribeOk;
self->state.set_value(as);
}
void on_unsubscribe_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
string msg = fmt::format("unsubscribe failed: {}", MQTTAsync_strerror(response ? response->code : 0));
spdlog::error(msg);
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::UnsubscribeFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_conn_fail(void* context, MQTTAsync_failureData* response)
{
MqttHelper *self = (MqttHelper *) context;
string msg = fmt::format("Connect failed: {}", MQTTAsync_strerror(response ? response->code : 0));
spdlog::error(msg);
//self->state.set_exception(StrException(msg));
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::ConnFail;
as.rc = response?response->code:0;
self->state.set_value(as);
}
void on_sent(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
spdlog::debug("Message delivery confirmed, tok: {}", response->token);
}
void on_connected(void* context, MQTTAsync_successData* response)
{
MqttHelper *self = (MqttHelper *) context;
MqttHelper::AsyncResult as;
as.state = MqttHelper::State::Ready;
self->state.set_value(as);
spdlog::info("connected to mqtt {}", self->addr);
}
#ifdef TEST_MQTT
#include <httplib.h>
void my_on_msg1(MqttHelper *hlp, const void * const data, int len, string topic)
{
string msg;
msg.resize(len+1);
memcpy((void*)&msg[0], data, len);
*(char*)(&msg[0] +len) = 0;
spdlog::info("I1 recv on {}: {}", topic, msg);
(*hlp)[to_string(stoi(topic) + 1)]("hello ack1", 11);
}
void my_on_msg2(MqttHelper *hlp, const void * const data, int len, string topic)
{
string msg;
msg.resize(len+1);
memcpy((void*)&msg[0], data, len);
*(char*)(&msg[0] +len) = 0;
spdlog::info("I2 recv on {}: {}", topic, msg);
(*hlp)[to_string(stoi(topic) + 1)]("hello ack2", 11);
}
return MqttMgr::insts[key];
}
int main()
{
string mqtt_url = "tcp://admin:vJ3zHqWrHbrqxVMT@evcloudsvc.ilabservice.cloud:11883";
string mqtt_cid = "testaabbc";
static void remove(string mqtt_url, string devsn){
auto key = mqtt_url + "/" + devsn;
lock_guard<mutex> lg(MqttMgr::mut);
if(MqttMgr::insts.count(key) == 0){
auto tmpEnv = getenv("MQTT_URL");
if(tmpEnv) {
mqtt_url = string(tmpEnv);
}else{
delete MqttMgr::insts[key];
MqttMgr::insts.erase(key);
}
}
tmpEnv = getenv("CID");
if(tmpEnv) {
mqtt_cid = string(tmpEnv);
MqttMgr(){}
~MqttMgr(){
for(auto &x:insts){
if(x.second){
delete x.second;
}
insts.erase(x.first);
}
}
spdlog::set_level(spdlog::level::debug);
{
try {
MqttHelper hlp(mqtt_url, mqtt_cid);
hlp["hello"].pub("hello", 6, 1);
hlp.subscribe("no_on_msg", NULL);
auto x = thread([&hlp] {
for(int i = 0; i < 20; i ++)
/// low level api for report and response
static void _report_response(MqttHelper *client, string topic, string message)
{
hlp.subscribe(to_string(i), my_on_msg1);
// auto client = MqttMgr::get_instance(mqtt_url, dev_sn);
if(client == nullptr) {
spdlog::error("invalid mqtt client {}");
return;
}
auto rc = (*client)[topic].pub(message.c_str(), message.size(), 2, false);
if(rc < 0) {
spdlog::error("failed to pub mqtt message on {}: {}", topic, message);
}
else {
spdlog::info("successfully pub message at {}:{}", topic, message);
}
}
hlp["hello"].pub("hello1", 7, 1);
});
auto y = thread([&hlp] {
for(int i = 0; i < 10; i ++)
/// having data field along with code and msg
/// signature {"rid", "code", "msg", "cata", "data":{}}
static void report_response_json(MqttHelper *client, string topic, json& message)
{
hlp.subscribe(to_string(i), my_on_msg2);
// time
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
message.insert_or_assign(consts::kMsgTime, ts);
// rid
auto rid = message["rid"].as<string>();
message.insert_or_assign(consts::kMsgRid, rid);
// type
if(rid.empty()) {
message.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
}
for(int i = 10; i < 30; i ++)
{
hlp.subscribe(to_string(i), my_on_msg2, true);
else {
message.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
}
hlp["hello"].pub("hello2", 7, 1);
});
x.join();
y.join();
this_thread::sleep_for(chrono::seconds(20));
// sn
message.insert_or_assign(consts::kMsgSn, client->id);
_report_response(client, topic, message.to_string());
}
catch(exception &e) {
spdlog::error("failed to connect mqtt: {}, {}", mqtt_url, e.what());
exit(1);
/// no data field. only code and msg field
static void report_response_args(MqttHelper *client, string topic, int code, string message, string cata, string rid, json &&data)
{
json js;
/// current timestamp
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
js.insert_or_assign(consts::kMsgTime, ts);
/// rid
js.insert_or_assign(consts::kMsgRid, rid);
/// type
if(rid.empty()) {
js.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
}
else {
js.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
}
js.insert_or_assign(consts::kMsgCata, cata);
///
js.insert_or_assign(consts::kMsgCode, code);
js.insert_or_assign(consts::kMsgMsg, message);
js.insert_or_assign(consts::kMsgSn, client->id);
this_thread::sleep_for(chrono::seconds(2));
return 0;
}
_report_response(client, topic, js.to_string());
}
#endif
static void report_response_args(MqttHelper *client, string topic, int code, string message, string cata, string rid, json &data)
{
json js;
/// current timestamp
auto ts = chrono::duration_cast<chrono::milliseconds>(chrono::system_clock::now().time_since_epoch()).count();
js.insert_or_assign(consts::kMsgTime, ts);
/// rid
js.insert_or_assign(consts::kMsgRid, rid);
/// type
if(rid.empty()) {
js.insert_or_assign(consts::kMsgType, consts::kMsgTypeReport);
}
else {
js.insert_or_assign(consts::kMsgType, consts::kMsgTypeResponse);
}
js.insert_or_assign(consts::kMsgCata, cata);
///
js.insert_or_assign(consts::kMsgCode, code);
js.insert_or_assign(consts::kMsgMsg, message);
js.insert_or_assign(consts::kMsgSn, client->id);
_report_response(client, topic, js.to_string());
}
};
#endif
\ No newline at end of file
vendor/shared/include/ghc/filesystem.hpp 0 → 100644
浏览文件 @ b92bb722
This source diff could not be displayed because it is too large. You can view the blob instead.
vendor/shared/include/ghc/fs_fwd.hpp 0 → 100644
浏览文件 @ b92bb722
//---------------------------------------------------------------------------------------
//
// ghc::filesystem - A C++17-like filesystem implementation for C++11/C++14
//
//---------------------------------------------------------------------------------------
//
// Copyright (c) 2018, Steffen Schümann <s.schuemann@pobox.com>
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification,
// are permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this
// list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors
// may be used to endorse or promote products derived from this software without
// specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
//---------------------------------------------------------------------------------------
// fs_fwd.hpp - The forwarding header for the header/implementation seperated usage of
// ghc::filesystem.
// This file can be include at any place, where ghc::filesystem api is needed while
// not bleeding implementation details (e.g. system includes) into the global namespace,
// as long as one cpp includes fs_impl.hpp to deliver the matching implementations.
//---------------------------------------------------------------------------------------
#ifndef GHC_FILESYSTEM_FWD_H
#define GHC_FILESYSTEM_FWD_H
#define GHC_FILESYSTEM_FWD
#include <ghc/filesystem.hpp>
#endif // GHC_FILESYSTEM_FWD_H
vendor/shared/include/ghc/fs_impl.hpp 0 → 100644
浏览文件 @ b92bb722
//---------------------------------------------------------------------------------------
//
// ghc::filesystem - A C++17-like filesystem implementation for C++11/C++14
//
//---------------------------------------------------------------------------------------
//
// Copyright (c) 2018, Steffen Schümann <s.schuemann@pobox.com>
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification,
// are permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this
// list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors
// may be used to endorse or promote products derived from this software without
// specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
//---------------------------------------------------------------------------------------
// fs_impl.hpp - The implementation header for the header/implementation seperated usage of
// ghc::filesystem.
// This file can be used to hide the implementation of ghc::filesystem into a single cpp.
// The cpp has to include this before including fs_fwd.hpp directly or via a different
// header to work.
//---------------------------------------------------------------------------------------
#define GHC_FILESYSTEM_IMPLEMENTATION
#include <ghc/filesystem.hpp>
vendor/shared/include/ghc/fs_std.hpp 0 → 100644
浏览文件 @ b92bb722
//---------------------------------------------------------------------------------------
//
// ghc::filesystem - A C++17-like filesystem implementation for C++11/C++14
//
//---------------------------------------------------------------------------------------
//
// Copyright (c) 2018, Steffen Schümann <s.schuemann@pobox.com>
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification,
// are permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this
// list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors
// may be used to endorse or promote products derived from this software without
// specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
//---------------------------------------------------------------------------------------
// fs_std.hpp - The dynamic switching header that includes std::filesystem if detected
// or ghc::filesystem if not, and makes the resulting API available in the
// namespace fs.
//---------------------------------------------------------------------------------------
#ifndef GHC_FILESYSTEM_STD_H
#if defined(__cplusplus) && __cplusplus >= 201703L && defined(__has_include)
#if __has_include(<filesystem>)
#define GHC_USE_STD_FS
#include <filesystem>
namespace fs {
using namespace std::filesystem;
using ifstream = std::ifstream;
using ofstream = std::ofstream;
using fstream = std::fstream;
}
#endif
#endif
#ifndef GHC_USE_STD_FS
#define GHC_WIN_WSTRING_STRING_TYPE
#include <ghc/filesystem.hpp>
namespace fs {
using namespace ghc::filesystem;
using ifstream = ghc::filesystem::ifstream;
using ofstream = ghc::filesystem::ofstream;
using fstream = ghc::filesystem::fstream;
}
#endif
#endif // GHC_FILESYSTEM_STD_H
vendor/shared/include/ghc/fs_std_fwd.hpp 0 → 100644
浏览文件 @ b92bb722
//---------------------------------------------------------------------------------------
//
// ghc::filesystem - A C++17-like filesystem implementation for C++11/C++14
//
//---------------------------------------------------------------------------------------
//
// Copyright (c) 2018, Steffen Schümann <s.schuemann@pobox.com>
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification,
// are permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this
// list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors
// may be used to endorse or promote products derived from this software without
// specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
//---------------------------------------------------------------------------------------
// fs_std_fwd.hpp - The forwarding header for the header/implementation seperated usage of
// ghc::filesystem that uses std::filesystem if it detects it.
// This file can be include at any place, where fs::filesystem api is needed while
// not bleeding implementation details (e.g. system includes) into the global namespace,
// as long as one cpp includes fs_std_impl.hpp to deliver the matching implementations.
//---------------------------------------------------------------------------------------
#ifndef GHC_FILESYSTEM_STD_FWD_H
#define GHC_FILESYSTEM_STD_FWD_H
#if defined(__cplusplus) && __cplusplus >= 201703L && defined(__has_include)
#if __has_include(<filesystem>)
#define GHC_USE_STD_FS
#include <filesystem>
namespace fs {
using namespace std::filesystem;
using ifstream = std::ifstream;
using ofstream = std::ofstream;
using fstream = std::fstream;
}
#endif
#endif
#ifndef GHC_USE_STD_FS
#define GHC_WIN_WSTRING_STRING_TYPE
#define GHC_FILESYSTEM_FWD
#include <ghc/filesystem.hpp>
namespace fs {
using namespace ghc::filesystem;
using ifstream = ghc::filesystem::ifstream;
using ofstream = ghc::filesystem::ofstream;
using fstream = ghc::filesystem::fstream;
}
#endif
#endif // GHC_FILESYSTEM_STD_FWD_H
vendor/shared/include/ghc/fs_std_impl.hpp 0 → 100644
浏览文件 @ b92bb722
//---------------------------------------------------------------------------------------
//
// ghc::filesystem - A C++17-like filesystem implementation for C++11/C++14
//
//---------------------------------------------------------------------------------------
//
// Copyright (c) 2018, Steffen Schümann <s.schuemann@pobox.com>
// All rights reserved.
//
// Redistribution and use in source and binary forms, with or without modification,
// are permitted provided that the following conditions are met:
//
// 1. Redistributions of source code must retain the above copyright notice, this
// list of conditions and the following disclaimer.
//
// 2. Redistributions in binary form must reproduce the above copyright notice,
// this list of conditions and the following disclaimer in the documentation
// and/or other materials provided with the distribution.
//
// 3. Neither the name of the copyright holder nor the names of its contributors
// may be used to endorse or promote products derived from this software without
// specific prior written permission.
//
// THIS SOFTWARE IS PROVIDED BY THE COPYRIGHT HOLDERS AND CONTRIBUTORS "AS IS" AND
// ANY EXPRESS OR IMPLIED WARRANTIES, INCLUDING, BUT NOT LIMITED TO, THE IMPLIED
// WARRANTIES OF MERCHANTABILITY AND FITNESS FOR A PARTICULAR PURPOSE ARE
// DISCLAIMED. IN NO EVENT SHALL THE COPYRIGHT HOLDER OR CONTRIBUTORS BE LIABLE
// FOR ANY DIRECT, INDIRECT, INCIDENTAL, SPECIAL, EXEMPLARY, OR CONSEQUENTIAL
// DAMAGES (INCLUDING, BUT NOT LIMITED TO, PROCUREMENT OF SUBSTITUTE GOODS OR
// SERVICES; LOSS OF USE, DATA, OR PROFITS; OR BUSINESS INTERRUPTION) HOWEVER
// CAUSED AND ON ANY THEORY OF LIABILITY, WHETHER IN CONTRACT, STRICT LIABILITY,
// OR TORT (INCLUDING NEGLIGENCE OR OTHERWISE) ARISING IN ANY WAY OUT OF THE USE
// OF THIS SOFTWARE, EVEN IF ADVISED OF THE POSSIBILITY OF SUCH DAMAGE.
//
//---------------------------------------------------------------------------------------
// fs_std_impl.hpp - The implementation header for the header/implementation seperated usage of
// ghc::filesystem that does nothing if std::filesystem is detected.
// This file can be used to hide the implementation of ghc::filesystem into a single cpp.
// The cpp has to include this before including fs_std_fwd.hpp directly or via a different
// header to work.
//---------------------------------------------------------------------------------------
#if defined(__cplusplus) && __cplusplus >= 201703L && defined(__has_include)
#if __has_include(<filesystem>)
#define GHC_USE_STD_FS
#endif
#endif
#ifndef GHC_USE_STD_FS
#define GHC_WIN_WSTRING_STRING_TYPE
#define GHC_FILESYSTEM_IMPLEMENTATION
#include <ghc/filesystem.hpp>
#endif
vendor/x64/include/zmq.h 0 → 100644
浏览文件 @ b92bb722
/*
Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
This file is part of libzmq, the ZeroMQ core engine in C++.
libzmq is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License (LGPL) as published
by the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
As a special exception, the Contributors give you permission to link
this library with independent modules to produce an executable,
regardless of the license terms of these independent modules, and to
copy and distribute the resulting executable under terms of your choice,
provided that you also meet, for each linked independent module, the
terms and conditions of the license of that module. An independent
module is a module which is not derived from or based on this library.
If you modify this library, you must extend this exception to your
version of the library.
libzmq is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*************************************************************************
NOTE to contributors. This file comprises the principal public contract
for ZeroMQ API users. Any change to this file supplied in a stable
release SHOULD not break existing applications.
In practice this means that the value of constants must not change, and
that old values may not be reused for new constants.
*************************************************************************
*/
#ifndef __ZMQ_H_INCLUDED__
#define __ZMQ_H_INCLUDED__
/* Version macros for compile-time API version detection */
#define ZMQ_VERSION_MAJOR 4
#define ZMQ_VERSION_MINOR 3
#define ZMQ_VERSION_PATCH 2
#define ZMQ_MAKE_VERSION(major, minor, patch) \
((major) *10000 + (minor) *100 + (patch))
#define ZMQ_VERSION \
ZMQ_MAKE_VERSION (ZMQ_VERSION_MAJOR, ZMQ_VERSION_MINOR, ZMQ_VERSION_PATCH)
#ifdef __cplusplus
extern "C" {
#endif
#if !defined _WIN32_WCE
#include <errno.h>
#endif
#include <stddef.h>
#include <stdio.h>
#if defined _WIN32
// Set target version to Windows Server 2008, Windows Vista or higher.
// Windows XP (0x0501) is supported but without client & server socket types.
#ifndef _WIN32_WINNT
#define _WIN32_WINNT 0x0600
#endif
#ifdef __MINGW32__
// Require Windows XP or higher with MinGW for getaddrinfo().
#if (_WIN32_WINNT >= 0x0501)
#else
#error You need at least Windows XP target
#endif
#endif
#include <winsock2.h>
#endif
/* Handle DSO symbol visibility */
#if defined _WIN32
#if defined ZMQ_STATIC
#define ZMQ_EXPORT
#elif defined DLL_EXPORT
#define ZMQ_EXPORT __declspec(dllexport)
#else
#define ZMQ_EXPORT __declspec(dllimport)
#endif
#else
#if defined __SUNPRO_C || defined __SUNPRO_CC
#define ZMQ_EXPORT __global
#elif (defined __GNUC__ && __GNUC__ >= 4) || defined __INTEL_COMPILER
#define ZMQ_EXPORT __attribute__ ((visibility ("default")))
#else
#define ZMQ_EXPORT
#endif
#endif
/* Define integer types needed for event interface */
#define ZMQ_DEFINED_STDINT 1
#if defined ZMQ_HAVE_SOLARIS || defined ZMQ_HAVE_OPENVMS
#include <inttypes.h>
#elif defined _MSC_VER && _MSC_VER < 1600
#ifndef uint64_t
typedef unsigned __int64 uint64_t;
#endif
#ifndef int32_t
typedef __int32 int32_t;
#endif
#ifndef uint32_t
typedef unsigned __int32 uint32_t;
#endif
#ifndef uint16_t
typedef unsigned __int16 uint16_t;
#endif
#ifndef uint8_t
typedef unsigned __int8 uint8_t;
#endif
#else
#include <stdint.h>
#endif
// 32-bit AIX's pollfd struct members are called reqevents and rtnevents so it
// defines compatibility macros for them. Need to include that header first to
// stop build failures since zmq_pollset_t defines them as events and revents.
#ifdef ZMQ_HAVE_AIX
#include <poll.h>
#endif
/******************************************************************************/
/* 0MQ errors. */
/******************************************************************************/
/* A number random enough not to collide with different errno ranges on */
/* different OSes. The assumption is that error_t is at least 32-bit type. */
#define ZMQ_HAUSNUMERO 156384712
/* On Windows platform some of the standard POSIX errnos are not defined. */
#ifndef ENOTSUP
#define ENOTSUP (ZMQ_HAUSNUMERO + 1)
#endif
#ifndef EPROTONOSUPPORT
#define EPROTONOSUPPORT (ZMQ_HAUSNUMERO + 2)
#endif
#ifndef ENOBUFS
#define ENOBUFS (ZMQ_HAUSNUMERO + 3)
#endif
#ifndef ENETDOWN
#define ENETDOWN (ZMQ_HAUSNUMERO + 4)
#endif
#ifndef EADDRINUSE
#define EADDRINUSE (ZMQ_HAUSNUMERO + 5)
#endif
#ifndef EADDRNOTAVAIL
#define EADDRNOTAVAIL (ZMQ_HAUSNUMERO + 6)
#endif
#ifndef ECONNREFUSED
#define ECONNREFUSED (ZMQ_HAUSNUMERO + 7)
#endif
#ifndef EINPROGRESS
#define EINPROGRESS (ZMQ_HAUSNUMERO + 8)
#endif
#ifndef ENOTSOCK
#define ENOTSOCK (ZMQ_HAUSNUMERO + 9)
#endif
#ifndef EMSGSIZE
#define EMSGSIZE (ZMQ_HAUSNUMERO + 10)
#endif
#ifndef EAFNOSUPPORT
#define EAFNOSUPPORT (ZMQ_HAUSNUMERO + 11)
#endif
#ifndef ENETUNREACH
#define ENETUNREACH (ZMQ_HAUSNUMERO + 12)
#endif
#ifndef ECONNABORTED
#define ECONNABORTED (ZMQ_HAUSNUMERO + 13)
#endif
#ifndef ECONNRESET
#define ECONNRESET (ZMQ_HAUSNUMERO + 14)
#endif
#ifndef ENOTCONN
#define ENOTCONN (ZMQ_HAUSNUMERO + 15)
#endif
#ifndef ETIMEDOUT
#define ETIMEDOUT (ZMQ_HAUSNUMERO + 16)
#endif
#ifndef EHOSTUNREACH
#define EHOSTUNREACH (ZMQ_HAUSNUMERO + 17)
#endif
#ifndef ENETRESET
#define ENETRESET (ZMQ_HAUSNUMERO + 18)
#endif
/* Native 0MQ error codes. */
#define EFSM (ZMQ_HAUSNUMERO + 51)
#define ENOCOMPATPROTO (ZMQ_HAUSNUMERO + 52)
#define ETERM (ZMQ_HAUSNUMERO + 53)
#define EMTHREAD (ZMQ_HAUSNUMERO + 54)
/* This function retrieves the errno as it is known to 0MQ library. The goal */
/* of this function is to make the code 100% portable, including where 0MQ */
/* compiled with certain CRT library (on Windows) is linked to an */
/* application that uses different CRT library. */
ZMQ_EXPORT int zmq_errno (void);
/* Resolves system errors and 0MQ errors to human-readable string. */
ZMQ_EXPORT const char *zmq_strerror (int errnum_);
/* Run-time API version detection */
ZMQ_EXPORT void zmq_version (int *major_, int *minor_, int *patch_);
/******************************************************************************/
/* 0MQ infrastructure (a.k.a. context) initialisation & termination. */
/******************************************************************************/
/* Context options */
#define ZMQ_IO_THREADS 1
#define ZMQ_MAX_SOCKETS 2
#define ZMQ_SOCKET_LIMIT 3
#define ZMQ_THREAD_PRIORITY 3
#define ZMQ_THREAD_SCHED_POLICY 4
#define ZMQ_MAX_MSGSZ 5
#define ZMQ_MSG_T_SIZE 6
#define ZMQ_THREAD_AFFINITY_CPU_ADD 7
#define ZMQ_THREAD_AFFINITY_CPU_REMOVE 8
#define ZMQ_THREAD_NAME_PREFIX 9
/* Default for new contexts */
#define ZMQ_IO_THREADS_DFLT 1
#define ZMQ_MAX_SOCKETS_DFLT 1023
#define ZMQ_THREAD_PRIORITY_DFLT -1
#define ZMQ_THREAD_SCHED_POLICY_DFLT -1
ZMQ_EXPORT void *zmq_ctx_new (void);
ZMQ_EXPORT int zmq_ctx_term (void *context_);
ZMQ_EXPORT int zmq_ctx_shutdown (void *context_);
ZMQ_EXPORT int zmq_ctx_set (void *context_, int option_, int optval_);
ZMQ_EXPORT int zmq_ctx_get (void *context_, int option_);
/* Old (legacy) API */
ZMQ_EXPORT void *zmq_init (int io_threads_);
ZMQ_EXPORT int zmq_term (void *context_);
ZMQ_EXPORT int zmq_ctx_destroy (void *context_);
/******************************************************************************/
/* 0MQ message definition. */
/******************************************************************************/
/* Some architectures, like sparc64 and some variants of aarch64, enforce pointer
* alignment and raise sigbus on violations. Make sure applications allocate
* zmq_msg_t on addresses aligned on a pointer-size boundary to avoid this issue.
*/
typedef struct zmq_msg_t
{
#if defined(_MSC_VER) && (defined(_M_X64) || defined(_M_ARM64))
__declspec(align (8)) unsigned char _[64];
#elif defined(_MSC_VER) && (defined(_M_IX86) || defined(_M_ARM_ARMV7VE))
__declspec(align (4)) unsigned char _[64];
#elif defined(__GNUC__) || defined(__INTEL_COMPILER) \
|| (defined(__SUNPRO_C) && __SUNPRO_C >= 0x590) \
|| (defined(__SUNPRO_CC) && __SUNPRO_CC >= 0x590)
unsigned char _[64] __attribute__ ((aligned (sizeof (void *))));
#else
unsigned char _[64];
#endif
} zmq_msg_t;
typedef void(zmq_free_fn) (void *data_, void *hint_);
ZMQ_EXPORT int zmq_msg_init (zmq_msg_t *msg_);
ZMQ_EXPORT int zmq_msg_init_size (zmq_msg_t *msg_, size_t size_);
ZMQ_EXPORT int zmq_msg_init_data (
zmq_msg_t *msg_, void *data_, size_t size_, zmq_free_fn *ffn_, void *hint_);
ZMQ_EXPORT int zmq_msg_send (zmq_msg_t *msg_, void *s_, int flags_);
ZMQ_EXPORT int zmq_msg_recv (zmq_msg_t *msg_, void *s_, int flags_);
ZMQ_EXPORT int zmq_msg_close (zmq_msg_t *msg_);
ZMQ_EXPORT int zmq_msg_move (zmq_msg_t *dest_, zmq_msg_t *src_);
ZMQ_EXPORT int zmq_msg_copy (zmq_msg_t *dest_, zmq_msg_t *src_);
ZMQ_EXPORT void *zmq_msg_data (zmq_msg_t *msg_);
ZMQ_EXPORT size_t zmq_msg_size (const zmq_msg_t *msg_);
ZMQ_EXPORT int zmq_msg_more (const zmq_msg_t *msg_);
ZMQ_EXPORT int zmq_msg_get (const zmq_msg_t *msg_, int property_);
ZMQ_EXPORT int zmq_msg_set (zmq_msg_t *msg_, int property_, int optval_);
ZMQ_EXPORT const char *zmq_msg_gets (const zmq_msg_t *msg_,
const char *property_);
/******************************************************************************/
/* 0MQ socket definition. */
/******************************************************************************/
/* Socket types. */
#define ZMQ_PAIR 0
#define ZMQ_PUB 1
#define ZMQ_SUB 2
#define ZMQ_REQ 3
#define ZMQ_REP 4
#define ZMQ_DEALER 5
#define ZMQ_ROUTER 6
#define ZMQ_PULL 7
#define ZMQ_PUSH 8
#define ZMQ_XPUB 9
#define ZMQ_XSUB 10
#define ZMQ_STREAM 11
/* Deprecated aliases */
#define ZMQ_XREQ ZMQ_DEALER
#define ZMQ_XREP ZMQ_ROUTER
/* Socket options. */
#define ZMQ_AFFINITY 4
#define ZMQ_ROUTING_ID 5
#define ZMQ_SUBSCRIBE 6
#define ZMQ_UNSUBSCRIBE 7
#define ZMQ_RATE 8
#define ZMQ_RECOVERY_IVL 9
#define ZMQ_SNDBUF 11
#define ZMQ_RCVBUF 12
#define ZMQ_RCVMORE 13
#define ZMQ_FD 14
#define ZMQ_EVENTS 15
#define ZMQ_TYPE 16
#define ZMQ_LINGER 17
#define ZMQ_RECONNECT_IVL 18
#define ZMQ_BACKLOG 19
#define ZMQ_RECONNECT_IVL_MAX 21
#define ZMQ_MAXMSGSIZE 22
#define ZMQ_SNDHWM 23
#define ZMQ_RCVHWM 24
#define ZMQ_MULTICAST_HOPS 25
#define ZMQ_RCVTIMEO 27
#define ZMQ_SNDTIMEO 28
#define ZMQ_LAST_ENDPOINT 32
#define ZMQ_ROUTER_MANDATORY 33
#define ZMQ_TCP_KEEPALIVE 34
#define ZMQ_TCP_KEEPALIVE_CNT 35
#define ZMQ_TCP_KEEPALIVE_IDLE 36
#define ZMQ_TCP_KEEPALIVE_INTVL 37
#define ZMQ_IMMEDIATE 39
#define ZMQ_XPUB_VERBOSE 40
#define ZMQ_ROUTER_RAW 41
#define ZMQ_IPV6 42
#define ZMQ_MECHANISM 43
#define ZMQ_PLAIN_SERVER 44
#define ZMQ_PLAIN_USERNAME 45
#define ZMQ_PLAIN_PASSWORD 46
#define ZMQ_CURVE_SERVER 47
#define ZMQ_CURVE_PUBLICKEY 48
#define ZMQ_CURVE_SECRETKEY 49
#define ZMQ_CURVE_SERVERKEY 50
#define ZMQ_PROBE_ROUTER 51
#define ZMQ_REQ_CORRELATE 52
#define ZMQ_REQ_RELAXED 53
#define ZMQ_CONFLATE 54
#define ZMQ_ZAP_DOMAIN 55
#define ZMQ_ROUTER_HANDOVER 56
#define ZMQ_TOS 57
#define ZMQ_CONNECT_ROUTING_ID 61
#define ZMQ_GSSAPI_SERVER 62
#define ZMQ_GSSAPI_PRINCIPAL 63
#define ZMQ_GSSAPI_SERVICE_PRINCIPAL 64
#define ZMQ_GSSAPI_PLAINTEXT 65
#define ZMQ_HANDSHAKE_IVL 66
#define ZMQ_SOCKS_PROXY 68
#define ZMQ_XPUB_NODROP 69
#define ZMQ_BLOCKY 70
#define ZMQ_XPUB_MANUAL 71
#define ZMQ_XPUB_WELCOME_MSG 72
#define ZMQ_STREAM_NOTIFY 73
#define ZMQ_INVERT_MATCHING 74
#define ZMQ_HEARTBEAT_IVL 75
#define ZMQ_HEARTBEAT_TTL 76
#define ZMQ_HEARTBEAT_TIMEOUT 77
#define ZMQ_XPUB_VERBOSER 78
#define ZMQ_CONNECT_TIMEOUT 79
#define ZMQ_TCP_MAXRT 80
#define ZMQ_THREAD_SAFE 81
#define ZMQ_MULTICAST_MAXTPDU 84
#define ZMQ_VMCI_BUFFER_SIZE 85
#define ZMQ_VMCI_BUFFER_MIN_SIZE 86
#define ZMQ_VMCI_BUFFER_MAX_SIZE 87
#define ZMQ_VMCI_CONNECT_TIMEOUT 88
#define ZMQ_USE_FD 89
#define ZMQ_GSSAPI_PRINCIPAL_NAMETYPE 90
#define ZMQ_GSSAPI_SERVICE_PRINCIPAL_NAMETYPE 91
#define ZMQ_BINDTODEVICE 92
/* Message options */
#define ZMQ_MORE 1
#define ZMQ_SHARED 3
/* Send/recv options. */
#define ZMQ_DONTWAIT 1
#define ZMQ_SNDMORE 2
/* Security mechanisms */
#define ZMQ_NULL 0
#define ZMQ_PLAIN 1
#define ZMQ_CURVE 2
#define ZMQ_GSSAPI 3
/* RADIO-DISH protocol */
#define ZMQ_GROUP_MAX_LENGTH 15
/* Deprecated options and aliases */
#define ZMQ_IDENTITY ZMQ_ROUTING_ID
#define ZMQ_CONNECT_RID ZMQ_CONNECT_ROUTING_ID
#define ZMQ_TCP_ACCEPT_FILTER 38
#define ZMQ_IPC_FILTER_PID 58
#define ZMQ_IPC_FILTER_UID 59
#define ZMQ_IPC_FILTER_GID 60
#define ZMQ_IPV4ONLY 31
#define ZMQ_DELAY_ATTACH_ON_CONNECT ZMQ_IMMEDIATE
#define ZMQ_NOBLOCK ZMQ_DONTWAIT
#define ZMQ_FAIL_UNROUTABLE ZMQ_ROUTER_MANDATORY
#define ZMQ_ROUTER_BEHAVIOR ZMQ_ROUTER_MANDATORY
/* Deprecated Message options */
#define ZMQ_SRCFD 2
/******************************************************************************/
/* GSSAPI definitions */
/******************************************************************************/
/* GSSAPI principal name types */
#define ZMQ_GSSAPI_NT_HOSTBASED 0
#define ZMQ_GSSAPI_NT_USER_NAME 1
#define ZMQ_GSSAPI_NT_KRB5_PRINCIPAL 2
/******************************************************************************/
/* 0MQ socket events and monitoring */
/******************************************************************************/
/* Socket transport events (TCP, IPC and TIPC only) */
#define ZMQ_EVENT_CONNECTED 0x0001
#define ZMQ_EVENT_CONNECT_DELAYED 0x0002
#define ZMQ_EVENT_CONNECT_RETRIED 0x0004
#define ZMQ_EVENT_LISTENING 0x0008
#define ZMQ_EVENT_BIND_FAILED 0x0010
#define ZMQ_EVENT_ACCEPTED 0x0020
#define ZMQ_EVENT_ACCEPT_FAILED 0x0040
#define ZMQ_EVENT_CLOSED 0x0080
#define ZMQ_EVENT_CLOSE_FAILED 0x0100
#define ZMQ_EVENT_DISCONNECTED 0x0200
#define ZMQ_EVENT_MONITOR_STOPPED 0x0400
#define ZMQ_EVENT_ALL 0xFFFF
/* Unspecified system errors during handshake. Event value is an errno. */
#define ZMQ_EVENT_HANDSHAKE_FAILED_NO_DETAIL 0x0800
/* Handshake complete successfully with successful authentication (if *
* enabled). Event value is unused. */
#define ZMQ_EVENT_HANDSHAKE_SUCCEEDED 0x1000
/* Protocol errors between ZMTP peers or between server and ZAP handler. *
* Event value is one of ZMQ_PROTOCOL_ERROR_* */
#define ZMQ_EVENT_HANDSHAKE_FAILED_PROTOCOL 0x2000
/* Failed authentication requests. Event value is the numeric ZAP status *
* code, i.e. 300, 400 or 500. */
#define ZMQ_EVENT_HANDSHAKE_FAILED_AUTH 0x4000
#define ZMQ_PROTOCOL_ERROR_ZMTP_UNSPECIFIED 0x10000000
#define ZMQ_PROTOCOL_ERROR_ZMTP_UNEXPECTED_COMMAND 0x10000001
#define ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_SEQUENCE 0x10000002
#define ZMQ_PROTOCOL_ERROR_ZMTP_KEY_EXCHANGE 0x10000003
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_UNSPECIFIED 0x10000011
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_MESSAGE 0x10000012
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_HELLO 0x10000013
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_INITIATE 0x10000014
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_ERROR 0x10000015
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_READY 0x10000016
#define ZMQ_PROTOCOL_ERROR_ZMTP_MALFORMED_COMMAND_WELCOME 0x10000017
#define ZMQ_PROTOCOL_ERROR_ZMTP_INVALID_METADATA 0x10000018
// the following two may be due to erroneous configuration of a peer
#define ZMQ_PROTOCOL_ERROR_ZMTP_CRYPTOGRAPHIC 0x11000001
#define ZMQ_PROTOCOL_ERROR_ZMTP_MECHANISM_MISMATCH 0x11000002
#define ZMQ_PROTOCOL_ERROR_ZAP_UNSPECIFIED 0x20000000
#define ZMQ_PROTOCOL_ERROR_ZAP_MALFORMED_REPLY 0x20000001
#define ZMQ_PROTOCOL_ERROR_ZAP_BAD_REQUEST_ID 0x20000002
#define ZMQ_PROTOCOL_ERROR_ZAP_BAD_VERSION 0x20000003
#define ZMQ_PROTOCOL_ERROR_ZAP_INVALID_STATUS_CODE 0x20000004
#define ZMQ_PROTOCOL_ERROR_ZAP_INVALID_METADATA 0x20000005
ZMQ_EXPORT void *zmq_socket (void *, int type_);
ZMQ_EXPORT int zmq_close (void *s_);
ZMQ_EXPORT int
zmq_setsockopt (void *s_, int option_, const void *optval_, size_t optvallen_);
ZMQ_EXPORT int
zmq_getsockopt (void *s_, int option_, void *optval_, size_t *optvallen_);
ZMQ_EXPORT int zmq_bind (void *s_, const char *addr_);
ZMQ_EXPORT int zmq_connect (void *s_, const char *addr_);
ZMQ_EXPORT int zmq_unbind (void *s_, const char *addr_);
ZMQ_EXPORT int zmq_disconnect (void *s_, const char *addr_);
ZMQ_EXPORT int zmq_send (void *s_, const void *buf_, size_t len_, int flags_);
ZMQ_EXPORT int
zmq_send_const (void *s_, const void *buf_, size_t len_, int flags_);
ZMQ_EXPORT int zmq_recv (void *s_, void *buf_, size_t len_, int flags_);
ZMQ_EXPORT int zmq_socket_monitor (void *s_, const char *addr_, int events_);
/******************************************************************************/
/* Deprecated I/O multiplexing. Prefer using zmq_poller API */
/******************************************************************************/
#define ZMQ_POLLIN 1
#define ZMQ_POLLOUT 2
#define ZMQ_POLLERR 4
#define ZMQ_POLLPRI 8
typedef struct zmq_pollitem_t
{
void *socket;
#if defined _WIN32
SOCKET fd;
#else
int fd;
#endif
short events;
short revents;
} zmq_pollitem_t;
#define ZMQ_POLLITEMS_DFLT 16
ZMQ_EXPORT int zmq_poll (zmq_pollitem_t *items_, int nitems_, long timeout_);
/******************************************************************************/
/* Message proxying */
/******************************************************************************/
ZMQ_EXPORT int zmq_proxy (void *frontend_, void *backend_, void *capture_);
ZMQ_EXPORT int zmq_proxy_steerable (void *frontend_,
void *backend_,
void *capture_,
void *control_);
/******************************************************************************/
/* Probe library capabilities */
/******************************************************************************/
#define ZMQ_HAS_CAPABILITIES 1
ZMQ_EXPORT int zmq_has (const char *capability_);
/* Deprecated aliases */
#define ZMQ_STREAMER 1
#define ZMQ_FORWARDER 2
#define ZMQ_QUEUE 3
/* Deprecated methods */
ZMQ_EXPORT int zmq_device (int type_, void *frontend_, void *backend_);
ZMQ_EXPORT int zmq_sendmsg (void *s_, zmq_msg_t *msg_, int flags_);
ZMQ_EXPORT int zmq_recvmsg (void *s_, zmq_msg_t *msg_, int flags_);
struct iovec;
ZMQ_EXPORT int
zmq_sendiov (void *s_, struct iovec *iov_, size_t count_, int flags_);
ZMQ_EXPORT int
zmq_recviov (void *s_, struct iovec *iov_, size_t *count_, int flags_);
/******************************************************************************/
/* Encryption functions */
/******************************************************************************/
/* Encode data with Z85 encoding. Returns encoded data */
ZMQ_EXPORT char *
zmq_z85_encode (char *dest_, const uint8_t *data_, size_t size_);
/* Decode data with Z85 encoding. Returns decoded data */
ZMQ_EXPORT uint8_t *zmq_z85_decode (uint8_t *dest_, const char *string_);
/* Generate z85-encoded public and private keypair with tweetnacl/libsodium. */
/* Returns 0 on success. */
ZMQ_EXPORT int zmq_curve_keypair (char *z85_public_key_, char *z85_secret_key_);
/* Derive the z85-encoded public key from the z85-encoded secret key. */
/* Returns 0 on success. */
ZMQ_EXPORT int zmq_curve_public (char *z85_public_key_,
const char *z85_secret_key_);
/******************************************************************************/
/* Atomic utility methods */
/******************************************************************************/
ZMQ_EXPORT void *zmq_atomic_counter_new (void);
ZMQ_EXPORT void zmq_atomic_counter_set (void *counter_, int value_);
ZMQ_EXPORT int zmq_atomic_counter_inc (void *counter_);
ZMQ_EXPORT int zmq_atomic_counter_dec (void *counter_);
ZMQ_EXPORT int zmq_atomic_counter_value (void *counter_);
ZMQ_EXPORT void zmq_atomic_counter_destroy (void **counter_p_);
/******************************************************************************/
/* Scheduling timers */
/******************************************************************************/
#define ZMQ_HAVE_TIMERS
typedef void(zmq_timer_fn) (int timer_id, void *arg);
ZMQ_EXPORT void *zmq_timers_new (void);
ZMQ_EXPORT int zmq_timers_destroy (void **timers_p);
ZMQ_EXPORT int
zmq_timers_add (void *timers, size_t interval, zmq_timer_fn handler, void *arg);
ZMQ_EXPORT int zmq_timers_cancel (void *timers, int timer_id);
ZMQ_EXPORT int
zmq_timers_set_interval (void *timers, int timer_id, size_t interval);
ZMQ_EXPORT int zmq_timers_reset (void *timers, int timer_id);
ZMQ_EXPORT long zmq_timers_timeout (void *timers);
ZMQ_EXPORT int zmq_timers_execute (void *timers);
/******************************************************************************/
/* These functions are not documented by man pages -- use at your own risk. */
/* If you need these to be part of the formal ZMQ API, then (a) write a man */
/* page, and (b) write a test case in tests. */
/******************************************************************************/
/* Helper functions are used by perf tests so that they don't have to care */
/* about minutiae of time-related functions on different OS platforms. */
/* Starts the stopwatch. Returns the handle to the watch. */
ZMQ_EXPORT void *zmq_stopwatch_start (void);
/* Returns the number of microseconds elapsed since the stopwatch was */
/* started, but does not stop or deallocate the stopwatch. */
ZMQ_EXPORT unsigned long zmq_stopwatch_intermediate (void *watch_);
/* Stops the stopwatch. Returns the number of microseconds elapsed since */
/* the stopwatch was started, and deallocates that watch. */
ZMQ_EXPORT unsigned long zmq_stopwatch_stop (void *watch_);
/* Sleeps for specified number of seconds. */
ZMQ_EXPORT void zmq_sleep (int seconds_);
typedef void(zmq_thread_fn) (void *);
/* Start a thread. Returns a handle to the thread. */
ZMQ_EXPORT void *zmq_threadstart (zmq_thread_fn *func_, void *arg_);
/* Wait for thread to complete then free up resources. */
ZMQ_EXPORT void zmq_threadclose (void *thread_);
/******************************************************************************/
/* These functions are DRAFT and disabled in stable releases, and subject to */
/* change at ANY time until declared stable. */
/******************************************************************************/
#ifdef ZMQ_BUILD_DRAFT_API
/* DRAFT Socket types. */
#define ZMQ_SERVER 12
#define ZMQ_CLIENT 13
#define ZMQ_RADIO 14
#define ZMQ_DISH 15
#define ZMQ_GATHER 16
#define ZMQ_SCATTER 17
#define ZMQ_DGRAM 18
/* DRAFT Socket options. */
#define ZMQ_ZAP_ENFORCE_DOMAIN 93
#define ZMQ_LOOPBACK_FASTPATH 94
#define ZMQ_METADATA 95
#define ZMQ_MULTICAST_LOOP 96
#define ZMQ_ROUTER_NOTIFY 97
#define ZMQ_XPUB_MANUAL_LAST_VALUE 98
#define ZMQ_SOCKS_USERNAME 99
#define ZMQ_SOCKS_PASSWORD 100
#define ZMQ_IN_BATCH_SIZE 101
#define ZMQ_OUT_BATCH_SIZE 102
/* DRAFT Context options */
#define ZMQ_ZERO_COPY_RECV 10
/* DRAFT Socket methods. */
ZMQ_EXPORT int zmq_join (void *s, const char *group);
ZMQ_EXPORT int zmq_leave (void *s, const char *group);
/* DRAFT Msg methods. */
ZMQ_EXPORT int zmq_msg_set_routing_id (zmq_msg_t *msg, uint32_t routing_id);
ZMQ_EXPORT uint32_t zmq_msg_routing_id (zmq_msg_t *msg);
ZMQ_EXPORT int zmq_msg_set_group (zmq_msg_t *msg, const char *group);
ZMQ_EXPORT const char *zmq_msg_group (zmq_msg_t *msg);
/* DRAFT Msg property names. */
#define ZMQ_MSG_PROPERTY_ROUTING_ID "Routing-Id"
#define ZMQ_MSG_PROPERTY_SOCKET_TYPE "Socket-Type"
#define ZMQ_MSG_PROPERTY_USER_ID "User-Id"
#define ZMQ_MSG_PROPERTY_PEER_ADDRESS "Peer-Address"
/* Router notify options */
#define ZMQ_NOTIFY_CONNECT 1
#define ZMQ_NOTIFY_DISCONNECT 2
/******************************************************************************/
/* Poller polling on sockets,fd and thread-safe sockets */
/******************************************************************************/
#define ZMQ_HAVE_POLLER
#if defined _WIN32
typedef SOCKET zmq_fd_t;
#else
typedef int zmq_fd_t;
#endif
typedef struct zmq_poller_event_t
{
void *socket;
zmq_fd_t fd;
void *user_data;
short events;
} zmq_poller_event_t;
ZMQ_EXPORT void *zmq_poller_new (void);
ZMQ_EXPORT int zmq_poller_destroy (void **poller_p);
ZMQ_EXPORT int
zmq_poller_add (void *poller, void *socket, void *user_data, short events);
ZMQ_EXPORT int zmq_poller_modify (void *poller, void *socket, short events);
ZMQ_EXPORT int zmq_poller_remove (void *poller, void *socket);
ZMQ_EXPORT int
zmq_poller_wait (void *poller, zmq_poller_event_t *event, long timeout);
ZMQ_EXPORT int zmq_poller_wait_all (void *poller,
zmq_poller_event_t *events,
int n_events,
long timeout);
ZMQ_EXPORT int zmq_poller_fd (void *poller, zmq_fd_t *fd);
ZMQ_EXPORT int
zmq_poller_add_fd (void *poller, zmq_fd_t fd, void *user_data, short events);
ZMQ_EXPORT int zmq_poller_modify_fd (void *poller, zmq_fd_t fd, short events);
ZMQ_EXPORT int zmq_poller_remove_fd (void *poller, zmq_fd_t fd);
ZMQ_EXPORT int zmq_socket_get_peer_state (void *socket,
const void *routing_id,
size_t routing_id_size);
/* DRAFT Socket monitoring events */
#define ZMQ_EVENT_PIPES_STATS 0x10000
#define ZMQ_CURRENT_EVENT_VERSION 1
#define ZMQ_CURRENT_EVENT_VERSION_DRAFT 2
#define ZMQ_EVENT_ALL_V1 ZMQ_EVENT_ALL
#define ZMQ_EVENT_ALL_V2 ZMQ_EVENT_ALL_V1 | ZMQ_EVENT_PIPES_STATS
ZMQ_EXPORT int zmq_socket_monitor_versioned (
void *s_, const char *addr_, uint64_t events_, int event_version_, int type_);
ZMQ_EXPORT int zmq_socket_monitor_pipes_stats (void *s);
#endif // ZMQ_BUILD_DRAFT_API
#undef ZMQ_EXPORT
#ifdef __cplusplus
}
#endif
#endif
vendor/x64/include/zmq_utils.h 0 → 100644
浏览文件 @ b92bb722
/*
Copyright (c) 2007-2016 Contributors as noted in the AUTHORS file
This file is part of libzmq, the ZeroMQ core engine in C++.
libzmq is free software; you can redistribute it and/or modify it under
the terms of the GNU Lesser General Public License (LGPL) as published
by the Free Software Foundation; either version 3 of the License, or
(at your option) any later version.
As a special exception, the Contributors give you permission to link
this library with independent modules to produce an executable,
regardless of the license terms of these independent modules, and to
copy and distribute the resulting executable under terms of your choice,
provided that you also meet, for each linked independent module, the
terms and conditions of the license of that module. An independent
module is a module which is not derived from or based on this library.
If you modify this library, you must extend this exception to your
version of the library.
libzmq is distributed in the hope that it will be useful, but WITHOUT
ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
FITNESS FOR A PARTICULAR PURPOSE. See the GNU Lesser General Public
License for more details.
You should have received a copy of the GNU Lesser General Public License
along with this program. If not, see <http://www.gnu.org/licenses/>.
*/
/* This file is deprecated, and all its functionality provided by zmq.h */
/* Note that -Wpedantic compilation requires GCC to avoid using its custom
extensions such as #warning, hence the trick below. Also, pragmas for
warnings or other messages are not standard, not portable, and not all
compilers even have an equivalent concept.
So in the worst case, this include file is treated as silently empty. */
#if defined(__clang__) || defined(__GNUC__) || defined(__GNUG__) \
|| defined(_MSC_VER)
#if defined(__GNUC__) || defined(__GNUG__)
#pragma GCC diagnostic push
#pragma GCC diagnostic warning "-Wcpp"
#pragma GCC diagnostic ignored "-Werror"
#pragma GCC diagnostic ignored "-Wall"
#endif
#pragma message( \
"Warning: zmq_utils.h is deprecated. All its functionality is provided by zmq.h.")
#if defined(__GNUC__) || defined(__GNUG__)
#pragma GCC diagnostic pop
#endif
#endif
vendor/x64/lib/libzmq.la 0 → 100755
浏览文件 @ b92bb722
# libzmq.la - a libtool library file
# Generated by libtool (GNU libtool) 2.4.6 Debian-2.4.6-2
#
# Please DO NOT delete this file!
# It is necessary for linking the library.
# The name that we can dlopen(3).
dlname='libzmq.so.5'
# Names of this library.
library_names='libzmq.so.5.2.2 libzmq.so.5 libzmq.so'
# The name of the static archive.
old_library='libzmq.a'
# Linker flags that cannot go in dependency_libs.
inherited_linker_flags=''
# Libraries that this one depends upon.
dependency_libs=' -lrt -lpthread'
# Names of additional weak libraries provided by this library
weak_library_names=''
# Version information for libzmq.
current=7
age=2
revision=2
# Is this an already installed library?
installed=yes
# Should we warn about portability when linking against -modules?
shouldnotlink=no
# Files to dlopen/dlpreopen
dlopen=''
dlpreopen=''
# Directory that this library needs to be installed in:
libdir='/mnt/shared/evcamera/vendor/libzmq../x64/lib'
vendor/x64/lib/pkgconfig/libzmq.pc 0 → 100644
浏览文件 @ b92bb722
prefix=/mnt/shared/evcamera/vendor/libzmq../x64
exec_prefix=${prefix}
libdir=${exec_prefix}/lib
includedir=${prefix}/include
Name: libzmq
Description: 0MQ c++ library
Version: 4.3.2
Libs: -L${libdir} -lzmq
Libs.private: -lstdc++ -lpthread -lrt
Requires.private:
Cflags: -I${includedir} -DZMQ_BUILD_DRAFT_API=1
Markdown 格式
0% 或
您添加了 0 人 到此讨论。请谨慎行事。
请先完成此评论的编辑！
请 注册 或者 后发表评论