Skip to content

M
mqtt-api
Commit 94d51d3f authored by zhoutotong's avatar zhoutotong
Browse files
Options

first commit

parents
Hide whitespace changes
Inline Side-by-side
Showing with 1062 additions and 0 deletions
.gitignore 0 → 100644
View file @ 94d51d3f
.vscode
build
\ No newline at end of file
CMakeLists.txt 0 → 100644
View file @ 94d51d3f
cmake_minimum_required(VERSION 3.5)
project(traffic_light_api)
set(CMAKE_CXX_STANDARD 14)
SET(CMAKE_CXX_STANDARD_REQUIRED ON)
SET(CMAKE_CXX_EXTENSIONS ON)
file(GLOB MSG_PROTOS ${CMAKE_CURRENT_SOURCE_DIR}/proto/*.proto)
set(PROTOS_DST_DIR ${CMAKE_CURRENT_BINARY_DIR}/proto)
message(STATUS "PROTOS_DST_DIR: ${PROTOS_DST_DIR}")
# 检查 PROTOS_DST_DIR 目录是否存在，不存在则创建
if(NOT EXISTS ${PROTOS_DST_DIR})
file(MAKE_DIRECTORY ${PROTOS_DST_DIR})
endif()
foreach(msg ${MSG_PROTOS})
message(STATUS "MSG_PROTOS: ${msg}")
get_filename_component(FIL_WE ${msg} NAME_WE)
execute_process(
COMMAND protoc -I ${CMAKE_CURRENT_SOURCE_DIR}/proto --cpp_out=${PROTOS_DST_DIR} ${msg}
)
endforeach()
add_subdirectory(${CMAKE_CURRENT_SOURCE_DIR}/demo)
# 收集源文件
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/mqtt SRCS)
AUX_SOURCE_DIRECTORY(${PROTOS_DST_DIR} SRCS)
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR}/api SRCS)
# 构建目标
add_library(${PROJECT_NAME} SHARED ${SRCS})
target_include_directories(${PROJECT_NAME}
PRIVATE
${CMAKE_CURRENT_SOURCE_DIR}
${CMAKE_CURRENT_SOURCE_DIR}/mqtt
${CMAKE_CURRENT_BINARY_DIR}/proto
)
target_link_libraries(${PROJECT_NAME}
pthread
protobuf
mosquitto
)
api/traffic_light.cpp 0 → 100644
View file @ 94d51d3f
#include "traffic_light.hpp"
/**
* @file traffic_light.cpp
* @author zhoutong (zhoutotong@qq.com)
* @brief
* @version 0.1
* @date 2024-08-21
*
* @copyright Copyright (c) 2024
*
*/
#include "api/traffic_light.hpp"
#include <chrono>
static uint64_t get_timestamp()
{
return std::chrono::duration_cast<std::chrono::milliseconds>(std::chrono::system_clock::now().time_since_epoch()).count();
}
TrafficLight::TrafficLight(const std::string &id, const double lat, const double lon, const double alt)
: id_(id)
, lat_(lat)
, lon_(lon)
, alt_(alt)
, cb_(nullptr)
, pub_(nullptr)
, sub_(nullptr)
{
}
TrafficLight::TrafficLight(const std::string &id, const ControlCallback& cb, const double lat, const double lon, const double alt)
: id_(id)
, lat_(lat)
, lon_(lon)
, alt_(alt)
, cb_(cb)
, pub_(std::make_shared<mqtt_dataio::publisher<fleet::RoadsideInfo>>("/traffic_light/" + id + "/status"))
, sub_(std::make_shared<mqtt_dataio::subscriber<fleet::LightControl>>("/traffic_light/" + id + "/control"
, std::bind(&TrafficLight::on_control_message, this, std::placeholders::_1)))
{
}
TrafficLight::~TrafficLight()
{
}
void TrafficLight::update(const fleet::LightStatus &s)
{
status_ = s;
if(pub_)
{
fleet::RoadsideInfo msg;
msg.set_id(id_);
msg.set_lightstatus(s);
msg.set_latitude(lat_);
msg.set_longitude(lon_);
msg.set_altitude(alt_);
pub_->publish(msg);
}
}
void TrafficLight::run_mqtt_client(const std::string &ip, const uint16_t &port, const std::string &user, const std::string &pwd)
{
mqtt_dataio::client::ins().init(ip, port, user, pwd);
}
void TrafficLight::on_control_message(const fleet::LightControl &msg)
{
if(cb_) cb_(id_, msg.lightstatus());
}
TrafficLightArray::TrafficLightArray(const std::string& topic_pub, const std::string& topic_control, const ControlCallback& cb)
: topic_pub_(topic_pub)
, topic_control_(topic_control)
, cb_(cb)
, pub_(std::make_shared<mqtt_dataio::publisher<fleet::RoadsideInfoList>>(topic_pub_))
, sub_(nullptr)
{
}
bool TrafficLightArray::add_light(const std::string &id, const double lat, const double lon, const double alt)
{
std::call_once(control_once_flag_, [this](){
sub_ = (std::make_shared<mqtt_dataio::subscriber<fleet::LightControl>>(
topic_control_, std::bind(&TrafficLightArray::on_control_message, this, std::placeholders::_1)));
});
auto itor = lights_.find(id);
if(itor != lights_.end()) return false;
lights_.emplace(id, std::make_shared<TrafficLight>(id, lat, lon, alt));
return true;
}
bool TrafficLightArray::remove_light(const std::string &id)
{
auto itor = lights_.find(id);
if(itor == lights_.end()) return false;
lights_.erase(itor);
return true;
}
bool TrafficLightArray::update(const std::string &id, const fleet::LightStatus &s)
{
auto itor = lights_.find(id);
if(itor == lights_.end()) return false;
itor->second->update(s);
return true;
}
void TrafficLightArray::update(const fleet::LightStatus &s)
{
for(auto itor = lights_.begin(); itor != lights_.end(); ++itor)
{
itor->second->update(s);
}
}
bool TrafficLightArray::publish()
{
if(pub_)
{
fleet::RoadsideInfoList msg;
for(auto itor = lights_.begin(); itor != lights_.end(); ++itor)
{
auto light = msg.add_roadsidelist();
light->set_id(itor->second->id_);
light->set_longitude(itor->second->lon_);
light->set_latitude(itor->second->lat_);
light->set_altitude(itor->second->alt_);
light->set_islight(1);
light->set_lightstatus(itor->second->status_);
}
pub_->publish(msg);
return true;
}
return false;
}
void TrafficLightArray::on_control_message(const fleet::LightControl &msg)
{
auto itor = lights_.find(msg.id());
if(itor != lights_.end() and cb_)
{
cb_(*itor->second, msg.lightstatus());
}
}
api/traffic_light.hpp 0 → 100644
View file @ 94d51d3f
/**
* @file traffic_light.hpp
* @author zhoutong (zhoutotong@qq.com)
* @brief
* @version 0.1
* @date 2024-08-21
*
* @copyright Copyright (c) 2024
*
*/
#pragma once
#include <mqtt/subcriber.hpp>
#include <mqtt/publisher.hpp>
#include <rs_info.pb.h>
class TrafficLight
{
friend class TrafficLightArray;
using ControlCallback = std::function<void(const std::string&, const fleet::LightStatus&)>;
template<typename T>
using ptr_t = std::shared_ptr<T>;
public:
TrafficLight(const std::string& id, const double lat = 0.0, const double lon = 0.0, const double alt = 0.0);
TrafficLight(const std::string& id, const ControlCallback& cb, const double lat = 0.0, const double lon = 0.0, const double alt = 0.0);
~TrafficLight();
void update(const fleet::LightStatus &s);
static void run_mqtt_client(const std::string& ip, const uint16_t& port, const std::string& user, const std::string& pwd);
private:
const std::string id_;
const double lat_;
const double lon_;
const double alt_;
fleet::LightStatus status_;
ControlCallback cb_;
ptr_t<mqtt_dataio::publisher<fleet::RoadsideInfo>> pub_;
ptr_t<mqtt_dataio::subscriber<fleet::LightControl>> sub_;
void on_control_message(const fleet::LightControl& msg);
}; // TrafficLight
class TrafficLightArray
{
template<typename T>
using ptr_t = std::shared_ptr<T>;
using ControlCallback = std::function<void(const TrafficLight&, const fleet::LightStatus&)>;
public:
TrafficLightArray(const std::string& topic_pub, const std::string& topic_control, const ControlCallback& cb = nullptr);
~TrafficLightArray() = default;
bool add_light(const std::string& id, const double lat = 0.0, const double lon = 0.0, const double alt = 0.0);
bool remove_light(const std::string& id);
bool update(const std::string& id, const fleet::LightStatus& s);
void update(const fleet::LightStatus& s);
bool publish();
private:
const std::string topic_pub_;
const std::string topic_control_;
ControlCallback cb_;
std::map<std::string, std::shared_ptr<TrafficLight>> lights_;
ptr_t<mqtt_dataio::publisher<fleet::RoadsideInfoList>> pub_;
ptr_t<mqtt_dataio::subscriber<fleet::LightControl>> sub_;
std::once_flag control_once_flag_;
void on_control_message(const fleet::LightControl& msg);
}; // TrafficLightArray
demo/CMakeLists.txt 0 → 100644
View file @ 94d51d3f
cmake_minimum_required(VERSION 3.5)
project(traffic_light_api_demo)
set(CMAKE_CXX_STANDARD 14)
SET(CMAKE_CXX_STANDARD_REQUIRED ON)
SET(CMAKE_CXX_EXTENSIONS ON)
# 收集源文件
AUX_SOURCE_DIRECTORY(${CMAKE_CURRENT_SOURCE_DIR} SRCS)
# 构建目标
add_executable(${PROJECT_NAME}-traffic_light traffic_light.cpp)
add_executable(${PROJECT_NAME}-multi_traffic_light multi_traffic_light.cpp)
target_include_directories(${PROJECT_NAME}-traffic_light
PRIVATE
${CMAKE_CURRENT_SOURCE_DIR}/..
${CMAKE_CURRENT_SOURCE_DIR}/../mqtt
${CMAKE_CURRENT_BINARY_DIR}/../proto
)
target_include_directories(${PROJECT_NAME}-multi_traffic_light
PRIVATE
${CMAKE_CURRENT_SOURCE_DIR}/..
${CMAKE_CURRENT_SOURCE_DIR}/../mqtt
${CMAKE_CURRENT_BINARY_DIR}/../proto
)
target_link_libraries(${PROJECT_NAME}-traffic_light
pthread
protobuf
mosquitto
traffic_light_api
)
target_link_libraries(${PROJECT_NAME}-multi_traffic_light
pthread
protobuf
mosquitto
traffic_light_api
)
\ No newline at end of file
demo/multi_traffic_light.cpp 0 → 100644
View file @ 94d51d3f
/**
* @file multi_traffic_light.cpp
* @author zhoutong (zhoutotong@qq.com)
* @brief
* @version 0.1
* @date 2024-08-21
*
* @copyright Copyright (c) 2024
*
*/
#include <api/traffic_light.hpp>
void on_control_traffic_light(const TrafficLight& light, const fleet::LightStatus& cmd)
{
// Do something with light and cmd
}
int main(int argc, char *argv[])
{
// Step0. 初始化 mqtt 客户端
TrafficLight::run_mqtt_client("127.0.0.1", 1883, "admin", "public");
// Step1. 创建 TrafficLightArray 对象
TrafficLightArray tla(
"/traffic_light/status",
"/traffic_light/control",
std::bind(on_control_traffic_light, std::placeholders::_1, std::placeholders::_2));
// Step2. 添加交通灯
tla.add_light("1"); // 无经纬度信息，默认为 0.0
tla.add_light("2"); // 无经纬度信息，默认为 0.0
tla.add_light("3", 39.123, 120.123, 50.0); // 指定经纬度信息
tla.add_light("4"); // 无经纬度信息，默认为 0.0
// Step3.1. 更新交通灯状态
tla.update("1", fleet::LightStatus::LIGHT_RED); // 更新 ID: 1 灯状态为 RED
tla.update("2", fleet::LightStatus::LIGHT_GREEN); // 更新 ID: 2 灯状态为 GREEN
tla.update("3", fleet::LightStatus::LIGHT_YELLOW); // 更新 ID: 3 灯状态为 YELLOW
tla.update("4", fleet::LightStatus::LIGHT_RED); // 更新 ID: 4 灯状态为 RED
// Step3.2. 同时更新多个交通灯状态
tla.update(fleet::LightStatus::LIGHT_RED); // 将当前全部灯状态更新为 RED
// Step4. 发布交通灯状态
tla.publish();
return 0;
}
demo/traffic_light.cpp 0 → 100644
View file @ 94d51d3f
/**
* @file traffic_light.cpp
* @author zhoutong (zhoutotong@qq.com)
* @brief
* @version 0.1
* @date 2024-08-21
*
* @copyright Copyright (c) 2024
*
*/
#include <api/traffic_light.hpp>
void on_control_traffic_light(const std::string& id, const fleet::LightStatus&)
{
// Do something with id and cmd
}
int main(int argc, char *argv[])
{
// Step0. 初始化 mqtt 客户端
TrafficLight::run_mqtt_client("127.0.0.1", 1883, "admin", "public");
// Step1. 创建 TrafficLightArray 对象
TrafficLight tl("1", std::bind(on_control_traffic_light, std::placeholders::_1, std::placeholders::_2));
// Step2. 更新交通灯状态
tl.update(fleet::LightStatus::LIGHT_RED); // 将当前全部灯状态更新为 RED, 并发布至 MQTT 服务器
return 0;
}
mqtt/mqtt_basic.hpp 0 → 100644
View file @ 94d51d3f
#pragma once
#include <vector>
#include <stdint.h>
namespace mqtt_dataio
{
class publisher_basic
{
public:
explicit publisher_basic() = default;
virtual ~publisher_basic() = default;
};
class subscriber_basic
{
public:
using DataBuffer = std::vector<uint8_t>;
explicit subscriber_basic() = default;
virtual ~subscriber_basic() = default;
virtual void on_message_arrival(const DataBuffer &) = 0;
};
} // namespace mqtt_dataio
mqtt/mqtt_client.cpp 0 → 100644
View file @ 94d51d3f
#include "mqtt_client.hpp"
namespace mqtt_dataio
{
bool client::__m_init = false;
struct mosquitto *client::__m_mosq = nullptr;
std::shared_ptr<std::thread> client::__m_thread = nullptr;
std::mutex client::__m_mtx;
std::map<std::string, mqtt_dataio::subscriber_basic *> client::__m_sub_list;
void client::init(const std::string &ip, const uint16_t &port, const std::string &user, const std::string &pwd)
{
if (__m_init)
return;
/* Required before calling other mosquitto functions */
mosquitto_lib_init();
/* Create a new client instance.
* id = NULL -> ask the broker to generate a client id for us
* clean session = true -> the broker should remove old sessions when we connect
* obj = NULL -> we aren't passing any of our private data for callbacks
*/
__m_mosq = mosquitto_new(nullptr, true, nullptr);
if (__m_mosq == nullptr)
{
std::cout << "Error: Out of memory." << std::endl;
return;
}
int rc;
rc = mosquitto_username_pw_set(__m_mosq, user.data(), pwd.data());
if (rc != MOSQ_ERR_SUCCESS)
{
std::cout << "Error: Problem setting username and password" << std::endl;
}
/* Configure callbacks. This should be done before connecting ideally. */
mosquitto_connect_callback_set(__m_mosq, &client::on_connect);
mosquitto_subscribe_callback_set(__m_mosq, &client::on_subscribe);
mosquitto_message_callback_set(__m_mosq, &client::on_message);
/* Connect to test.mosquitto.org on port 1883, with a keepalive of 60 seconds.
* This call makes the socket connection only, it does not complete the MQTT
* CONNECT/CONNACK flow, you should use mosquitto_loop_start() or
* mosquitto_loop_forever() for processing net traffic. */
rc = mosquitto_connect(__m_mosq, ip.data(), port, 60);
if (rc != MOSQ_ERR_SUCCESS)
{
mosquitto_destroy(__m_mosq);
std::cout << "Error: " << mosquitto_strerror(rc) << std::endl;
__m_init = false;
return;
}
__m_thread = std::make_shared<std::thread>([=]()
{ mosquitto_loop_forever(__m_mosq, -1, 1); });
__m_thread->detach();
{
std::lock_guard<std::mutex> lck(__m_mtx);
__m_init = true;
}
std::cout << "mqtt client init done, connecting to " << ip << ":" << port << std::endl;
}
void client::add_suber(const std::string &topic, mqtt_dataio::subscriber_basic *ptr)
{
{
std::lock_guard<std::mutex> lck(__m_mtx);
if (__m_sub_list.find(topic) == __m_sub_list.end())
{
__m_sub_list[topic] = ptr;
std::cout << "create mqtt_subcriber topic: " << topic << std::endl;
}
else
{
std::cout << "create mqtt_subcriber failed, had exist same topic: " << topic << std::endl;
}
}
int rc = mosquitto_subscribe(__m_mosq, NULL, topic.data(), 1);
if (rc != MOSQ_ERR_SUCCESS)
{
std::cout << "Error subscribing: " << mosquitto_strerror(rc) << std::endl;
/* We might as well disconnect if we were unable to subscribe */
// mosquitto_disconnect(__m_mosq);
}
}
void client::rm_suber(const std::string &topic)
{
{
std::lock_guard<std::mutex> lck(__m_mtx);
if (__m_sub_list.find(topic) != __m_sub_list.end())
{
__m_sub_list.erase(topic);
std::cout << "destroy mqtt_subcriber topic: " << topic << std::endl;
}
else
{
std::cout << "destroy mqtt_subcriber failed, not exist same topic: " << topic << std::endl;
}
}
}
void client::publish(const std::string &topic, const std::vector<uint8_t> &buff)
{
if (!__m_init)
return;
int rc = mosquitto_publish(__m_mosq, NULL, topic.data(), buff.size(), buff.data(), 1, false);
if (rc != MOSQ_ERR_SUCCESS)
{
std::cout << "Error publishing: " << mosquitto_strerror(rc) << std::endl;
}
}
client::~client()
{
mosquitto_disconnect(__m_mosq);
mosquitto_lib_cleanup();
__m_thread = nullptr;
{
std::lock_guard<std::mutex> lck(__m_mtx);
__m_sub_list.clear();
__m_init = false;
}
}
void client::on_connect(struct mosquitto *mosq, void *obj, int reason_code)
{
std::cout << "on_connect: " << mosquitto_connack_string(reason_code) << std::endl;
if (reason_code != 0)
{
/* If the connection fails for any reason, we don't want to keep on
* retrying in this example, so disconnect. Without this, the client
* will attempt to reconnect. */
// mosquitto_disconnect(mosq);
}
// /* Making subscriptions in the on_connect() callback means that if the
// * connection drops and is automatically resumed by the client, then the
// * subscriptions will be recreated when the client reconnects. */
// rc = mosquitto_subscribe(mosq, NULL, "example/temperature", 1);
// if(rc != MOSQ_ERR_SUCCESS){
// fprintf(stderr, "Error subscribing: %s\n", mosquitto_strerror(rc));
// /* We might as well disconnect if we were unable to subscribe */
// mosquitto_disconnect(mosq);
// }
}
/* Callback called when the broker sends a SUBACK in response to a SUBSCRIBE. */
void client::on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos)
{
int i;
bool have_subscription = false;
/* In this example we only subscribe to a single topic at once, but a
* SUBSCRIBE can contain many topics at once, so this is one way to check
* them all. */
for (i = 0; i < qos_count; i++)
{
std::cout << "on_subscribe: " << i << " :granted qos = " << granted_qos[i] << std::endl;
if (granted_qos[i] <= 2)
{
have_subscription = true;
}
}
if (have_subscription == false)
{
/* The broker rejected all of our subscriptions, we know we only sent
* the one SUBSCRIBE, so there is no point remaining connected. */
std::cout << "Error: All subscriptions rejected." << std::endl;
// mosquitto_disconnect(mosq);
}
}
void client::on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg)
{
std::string topic(msg->topic);
{
std::lock_guard<std::mutex> lck(__m_mtx);
auto it = __m_sub_list.find(topic);
if(it != __m_sub_list.end())
{
std::vector<uint8_t> data;
data.resize(msg->payloadlen);
memcpy(data.data(), msg->payload, msg->payloadlen);
it->second->on_message_arrival(data);
}
}
}
}
\ No newline at end of file
mqtt/mqtt_client.hpp 0 → 100644
View file @ 94d51d3f
#pragma once
#include "mqtt_basic.hpp"
#include <mosquitto.h>
#include <stdio.h>
#include <stdlib.h>
#include <string.h>
#include <unistd.h>
#include <iostream>
#include <map>
#include <memory>
#include <functional>
#include <thread>
#include <mutex>
namespace mqtt_dataio
{
class client
{
public:
static client &ins()
{
static client ins;
return ins;
}
void init(const std::string &ip, const uint16_t &port, const std::string &user, const std::string &pwd);
void add_suber(const std::string &topic, mqtt_dataio::subscriber_basic *ptr);
void rm_suber(const std::string &topic);
void publish(const std::string &topic, const std::vector<uint8_t> &buff);
~client();
private:
client() = default;
static void on_connect(struct mosquitto *mosq, void *obj, int reason_code);
static void on_subscribe(struct mosquitto *mosq, void *obj, int mid, int qos_count, const int *granted_qos);
static void on_message(struct mosquitto *mosq, void *obj, const struct mosquitto_message *msg);
private:
static bool __m_init;
static struct mosquitto *__m_mosq;
static std::shared_ptr<std::thread> __m_thread;
static std::mutex __m_mtx;
static std::map<std::string, mqtt_dataio::subscriber_basic *> __m_sub_list;
};
}
\ No newline at end of file
mqtt/publisher.hpp 0 → 100644
View file @ 94d51d3f
#pragma once
#include "mqtt_basic.hpp"
#include "mqtt_client.hpp"
#include <google/protobuf/message_lite.h>
namespace mqtt_dataio
{
template <typename T>
class publisher : public publisher_basic
{
public:
publisher() = delete;
publisher(const publisher &) = delete;
publisher(const std::string &name)
: __m_name(name)
{
std::cout << "create publisher topic: " << __m_name << std::endl;
}
~publisher() override
{
std::cout << "destroy publisher topic: " << __m_name << std::endl;
}
// void publish(const T &msg)
// {
// if constexpr (std::is_base_of<google::protobuf::MessageLite, T>::value)
// {
// std::string data;
// if (msg.SerializeToString(&data))
// {
// std::vector<uint8_t> buff(data.begin(), data.end());
// mqtt_dataio::client::ins().publish(__m_name, buff);
// }
// else
// {
// std::cout << "SerializeToString failed!!!" << std::endl;
// }
// }
// if constexpr (!std::is_base_of<google::protobuf::MessageLite, T>::value)
// {
// mqtt_dataio::client::ins().publish(__m_name, msg);
// }
// }
// std::enable_if<!std::is_base_of<google::protobuf::MessageLite, T>::value, const T &>::type
// typename std::enable_if<!std::is_base_of<google::protobuf::MessageLite, T>::value>::type
template<typename U = T, typename std::enable_if<!std::is_base_of<google::protobuf::MessageLite, U>::value, int>::type = 0>
void publish(const T& msg)
{
mqtt_dataio::client::ins().publish(__m_name, msg);
}
// std::enable_if<std::is_base_of<google::protobuf::MessageLite, T>::value, const T &>::type
// typename std::enable_if<std::is_base_of<google::protobuf::MessageLite, T>::value>::type
template<typename U = T, typename std::enable_if<std::is_base_of<google::protobuf::MessageLite, U>::value, int>::type = 0>
void publish(const google::protobuf::MessageLite& msg)
{
std::string data;
if (msg.SerializeToString(&data))
{
std::vector<uint8_t> buff(data.begin(), data.end());
mqtt_dataio::client::ins().publish(__m_name, buff);
}
else
{
std::cout << "SerializeToString failed!!!" << std::endl;
}
}
private:
std::string __m_name;
};
}
mqtt/subcriber.hpp 0 → 100644
View file @ 94d51d3f
#include "mqtt_basic.hpp"
#include "mqtt_client.hpp"
#include <string>
#include <google/protobuf/message_lite.h>
namespace mqtt_dataio
{
template <typename RT>
class subscriber : public subscriber_basic
{
public:
using DataCallback = std::function<void(const RT &)>;
public:
subscriber(const subscriber &) = delete;
subscriber &operator=(const subscriber &) = delete;
subscriber() = delete;
subscriber(const std::string &name, DataCallback cb)
: __m_name(name),
__m_cb(cb)
{
mqtt_dataio::client::ins().add_suber(__m_name, this);
}
~subscriber() override
{
mqtt_dataio::client::ins().rm_suber(__m_name);
}
virtual void on_message_arrival(const DataBuffer &buff) override
{
if (__m_cb) on_message_arrival_(buff);
}
private:
template<typename U = RT, typename std::enable_if<!std::is_base_of<google::protobuf::MessageLite, U>::value, int>::type = 0>
void on_message_arrival_(const DataBuffer& buff)
{
__m_cb(buff);
}
template<typename U = RT, typename std::enable_if<std::is_base_of<google::protobuf::MessageLite, U>::value, int>::type = 0>
void on_message_arrival_(const DataBuffer& buff)
{
RT msg;
msg.ParseFromArray(buff.data(), buff.size());
__m_cb(msg);
}
private:
std::string __m_name;
DataCallback __m_cb;
};
} // namespace mqtt_dataio
proto/monitor_info.proto 0 → 100755
View file @ 94d51d3f
syntax = "proto3";
package fleet;
// 路侧设备将实时数据上报调度平台。包括字段：
// ①设备ID
// ②经纬度坐标：精确到cm
// ③设备状态：在线、离线、故障
// ④实时电量：实时电量，单位%
enum RoadsideStatus
{
UNKNOWN_STATUS = 0; // 未知
ONLINE = 1; // 在线
OFFLINE = 2; // 离线
FAULT = 3; // 故障
}
message MonitorInfo
{
string roadside_id = 1; // 路侧设备ID
double longitude = 2; // 经度，单位cm
double latitude = 3; // 纬度，单位cm
uint64 timestamp = 4; // 时间戳，单位ms
RoadsideStatus status = 5; // 路侧设备状态
float battery = 6; // 实时电量，单位%
}
\ No newline at end of file
proto/rs_info.proto 0 → 100755
View file @ 94d51d3f
syntax = "proto3";
package fleet;
// 路侧信息
// ①设备ID：系统生成，唯一标识；
// ②经度：精确到cm；
// ③纬度：精确到cm；
// ④海拔：单位m，保留2位小数。
message RoadsideInfo {
string id = 1; // 交通灯id
double longitude = 2; // 经度
double latitude = 3; // 纬度
double altitude = 4; // 海拔
uint32 isLight = 5; // 是否有红绿灯
LightStatus lightStatus = 6; // 红绿灯的状态
}
enum LightStatus {
UNKNOWN_STATUS_1 = 0; // 未定义
LIGHT_GREEN = 1; // 绿灯
LIGHT_RED = 2; // 红灯
LIGHT_YELLOW = 3; // 黄灯
}
message LightControl {
string id = 1; // 路侧ID
LightStatus lightStatus = 2; // 变更颜色
uint64 timestamp = 3; // 当前时间戳
}
message RoadsideInfoList {
repeated RoadsideInfo roadsideList = 1;
}
proto/sdf.proto 0 → 100644
View file @ 94d51d3f
syntax = "proto3";
package SDF;
message ADF_EventTimeInfo{
uint64 startTime = 1; // 事件开始时间 #增改
uint64 endTime = 2; //事件结束事件 #增改
float startTimeConfidence = 3; //开始时间置信度 #增删改
float endTimeConfidence = 4; //结束时间置信度 #增删改
uint64 updateTime = 5; //最近更新时间 #仅由系统内部写入
}
message ADF_EventInfoSource{
/*
从右到左,每一个信息源类型占用2个Bit:
00: 状态未知, 01: 无该信息源, 10: 该信息源存在但无上报, 11: 该信息源存在并使用中.
0-1bit 移动终端上报
2-3bit 路侧端上报
4-5bit 车端上报
6-7bit 云端服务上报
8-9bit 云端用户上报
10-11bit 特殊用户上报
*/
uint64 sourceType = 1;//事件来源
string sourceTerminalID = 2; // 上报端 - 移动终端ID
string sourceVehicleID = 3; // 上报端 - 车端ID
string sourceRoadSideID = 4; // 上报端 - 路侧端ID
string sourceProviderID = 5; // 上报端 - 云端服务ID
string sourceUserID = 6; // 上报端 - 用户ID
}
// 描述交通信号灯的实时灯色，灯色剩余时长的数据结构 #增删改
message ADF_TrafficLightPhaseStage{
enum TrafficLightColorType{
UNKNOWN = 0; // 未知状态（离线）
BLACK = 1; // 黑灯 （不可用/故障）
RED = 2; // 红灯 （目前不可通行）
RED_FLASH = 3; // 闪烁红灯
YELLOW = 4; // 黄灯
YELLOW_FLASH = 5; //闪烁黄灯
GREEN = 6; // 绿灯
GREEN_FLASH = 7; // 闪烁绿灯
}
TrafficLightColorType lightColor = 1; //当前灯色
uint64 switchingTime = 2; // 灯色切换时间戳，单位秒。若为0，默认开始时间+持续600s。
TrafficLightColorType nextColor = 3; // 下一个灯色
}
//信号灯实时相位信息 #增删改
message EDF_TrafficLightRealTimeInfo{
string eventID = 1; // 事件ID #仅由系统分配
string trafficLightID = 2; // 关联信号灯ID #整体增写入时必选
string lightBulbID = 3; // 关联信号灯灯泡ID #整体增写入时可选
ADF_EventTimeInfo timeInfo = 4; //时间信息,仅使用其中的"最近更新时间字段" #仅由系统更新
ADF_TrafficLightPhaseStage phaseStage = 5; //实时灯色, 灯色剩余时长及变更计划
ADF_EventInfoSource eventInfoSource = 6; //事件信息来源 #仅由系统写入
}
//动态图层信息
message TrafficLightLayer
{
repeated EDF_TrafficLightRealTimeInfo trafficLightRealTimeInfoLayer = 1;
}
message RequestSequence
{
string requestID = 1; // 服务器分配的ID 和eventID相同 // 仅add 有
repeated EDF_TrafficLightRealTimeInfo listOfRequest = 2;
}
message TrafficLightLayerResponse
{
int32 errorCode = 1;
RequestSequence requestSequence = 2;
}
\ No newline at end of file
proto/trafficlight_cloud.proto 0 → 100644
View file @ 94d51d3f
syntax = "proto3";
package cloud;
//信号灯路口通行模式
message IntersectionTrafficMode {
enum TrafficMode {
UNKNOWN_MODE = 0; //未知
MANNED_MODE = 1; //有人车通行模式
DRIVERLESS_MODE = 2; //无人车通行模式
NONE_VEHICLE_MODE = 3; //无车通行模式
}
TrafficMode traffic_mode = 1; //当前的通行模式
uint64 update_time = 2; //通行模式最近更新时间戳，单位ms
}
readme.md 0 → 100644
View file @ 94d51d3f
# TrafficLight API
TrafficLight API 提供将红绿灯状态通过 MQTT 推送给平台的功能，同时可接收从服务端订阅灯的状态控制命令的功能。
## 1. 安装
1. 安装依赖
```bash
sudo apt install libprotobuf-dev protobuf-compiler
sudo apt install libmosquitto-dev
```
2. 集成代码
通过源码的方式集成到项目中，将本仓库代码完整拷贝至项目目录下，例如项目主目录。
在 CMakeLists.txt 文件中添加以下内容
```cmake
add_subdirectory(${CMAKE_SOURCE_DIR}/mqtt-api)
```
为目标添加库依赖
```cmake
target_link_libraries(<NAME-OF-TARGET>
...
other deps
...
pthread
protobuf
mosquitto
traffic_light_api
)
```
## 2. 接口说明
- 单灯控制接口示例
```c++
#include <api/traffic_light.hpp>
void on_control_traffic_light(const std::string& id, const fleet::LightStatus&)
{
// Do something with id and cmd
}
int main(int argc, char *argv[])
{
// Step0. 初始化 mqtt 客户端
TrafficLight::run_mqtt_client("127.0.0.1", 1883, "admin", "public");
// Step1. 创建 TrafficLightArray 对象
TrafficLight tl("1", std::bind(on_control_traffic_light, std::placeholders::_1, std::placeholders::_2));
// Step2. 更新交通灯状态
tl.update(fleet::LightStatus::LIGHT_RED); // 将当前全部灯状态更新为 RED, 并发布至 MQTT 服务器
return 0;
}
```
- 多灯控制接口示例
```c++
#include <api/traffic_light.hpp>
void on_control_traffic_light(const TrafficLight& light, const fleet::LightStatus& cmd)
{
// Do something with light and cmd
}
int main(int argc, char *argv[])
{
// Step0. 初始化 mqtt 客户端
TrafficLight::run_mqtt_client("127.0.0.1", 1883, "admin", "public");
// Step1. 创建 TrafficLightArray 对象
TrafficLightArray tla(
"/traffic_light/status",
"/traffic_light/control",
std::bind(on_control_traffic_light, std::placeholders::_1, std::placeholders::_2));
// Step2. 添加交通灯
tla.add_light("1"); // 无经纬度信息，默认为 0.0
tla.add_light("2"); // 无经纬度信息，默认为 0.0
tla.add_light("3", 39.123, 120.123, 50.0); // 指定经纬度信息
tla.add_light("4"); // 无经纬度信息，默认为 0.0
// Step3.1. 更新交通灯状态
tla.update("1", fleet::LightStatus::LIGHT_RED); // 更新 ID: 1 灯状态为 RED
tla.update("2", fleet::LightStatus::LIGHT_GREEN); // 更新 ID: 2 灯状态为 GREEN
tla.update("3", fleet::LightStatus::LIGHT_YELLOW); // 更新 ID: 3 灯状态为 YELLOW
tla.update("4", fleet::LightStatus::LIGHT_RED); // 更新 ID: 4 灯状态为 RED
// Step3.2. 同时更新多个交通灯状态
tla.update(fleet::LightStatus::LIGHT_RED); // 将当前全部灯状态更新为 RED
// Step4. 发布交通灯状态
tla.publish();
return 0;
}
```
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment