flowzl/softbus_daemon
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

4 division

Browse Source
This commit is contained in:
flower_linux
2026-05-11 19:17:42 +08:00
parent 7280fca08d
commit df5eb5529c
1385 changed files with 441353 additions and 1232 deletions
Download Patch File Download Diff File Expand all files Collapse all files

197
src/message_bus/ipc/DomEvent.cpp Normal file
View File

@@ -0,0 +1,197 @@
#include "message_bus/ipc/DomEvent.h"
#include <QDateTime>
#include <QJsonArray>
#include <QUuid>
namespace softbus::message_bus::ipc
{
namespace
{
QString qualityToString(softbus::core::models::DataQuality q)
{
using softbus::core::models::DataQuality;
switch (q) {
case DataQuality::GOOD:
return QStringLiteral("GOOD");
case DataQuality::BAD:
return QStringLiteral("BAD");
case DataQuality::UNCERTAIN:
return QStringLiteral("UNCERTAIN");
case DataQuality::STALE:
return QStringLiteral("STALE");
}
return QStringLiteral("BAD");
}
softbus::core::models::DataQuality qualityFromString(const QString& v)
{
using softbus::core::models::DataQuality;
const QString x = v.trimmed().toUpper();
if (x == QLatin1String("GOOD")) {
return DataQuality::GOOD;
}
if (x == QLatin1String("UNCERTAIN")) {
return DataQuality::UNCERTAIN;
}
if (x == QLatin1String("STALE")) {
return DataQuality::STALE;
}
return DataQuality::BAD;
}
QJsonObject domValueToJson(const softbus::core::models::DOMValue& v)
{
QJsonObject out;
std::visit(
[&out](const auto& x) {
using T = std::decay_t<decltype(x)>;
if constexpr (std::is_same_v<T, bool>) {
out.insert(QStringLiteral("type"), QStringLiteral("bool"));
out.insert(QStringLiteral("value"), x);
} else if constexpr (std::is_same_v<T, std::int32_t>) {
out.insert(QStringLiteral("type"), QStringLiteral("int32"));
out.insert(QStringLiteral("value"), static_cast<qint64>(x));
} else if constexpr (std::is_same_v<T, std::uint32_t>) {
out.insert(QStringLiteral("type"), QStringLiteral("uint32"));
out.insert(QStringLiteral("value"), static_cast<qint64>(x));
} else if constexpr (std::is_same_v<T, std::int64_t>) {
out.insert(QStringLiteral("type"), QStringLiteral("int64"));
out.insert(QStringLiteral("value"), static_cast<qint64>(x));
} else if constexpr (std::is_same_v<T, float>) {
out.insert(QStringLiteral("type"), QStringLiteral("float32"));
out.insert(QStringLiteral("value"), static_cast<double>(x));
} else {
out.insert(QStringLiteral("type"), QStringLiteral("float64"));
out.insert(QStringLiteral("value"), static_cast<double>(x));
}
},
v);
return out;
}
softbus::core::models::DOMValue domValueFromJson(const QJsonObject& obj)
{
const QString type = obj.value(QStringLiteral("type")).toString().trimmed().toLower();
const QJsonValue value = obj.value(QStringLiteral("value"));
if (type == QLatin1String("bool")) {
return softbus::core::models::DOMValue{value.toBool(false)};
}
if (type == QLatin1String("int32")) {
return softbus::core::models::DOMValue{static_cast<std::int32_t>(value.toInt(0))};
}
if (type == QLatin1String("uint32")) {
return softbus::core::models::DOMValue{static_cast<std::uint32_t>(value.toInt(0))};
}
if (type == QLatin1String("int64")) {
return softbus::core::models::DOMValue{static_cast<std::int64_t>(value.toVariant().toLongLong())};
}
if (type == QLatin1String("float32")) {
return softbus::core::models::DOMValue{static_cast<float>(value.toDouble(0.0))};
}
return softbus::core::models::DOMValue{static_cast<double>(value.toDouble(0.0))};
}
} // namespace
QJsonObject domMessageToJson(const softbus::core::models::DOMMessage& message)
{
QJsonObject out;
out.insert(QStringLiteral("messageId"), message.messageId);
out.insert(QStringLiteral("objectType"), message.objectType);
out.insert(QStringLiteral("objectRef"), message.objectRef);
out.insert(QStringLiteral("domainKey"), message.domainKey);
out.insert(QStringLiteral("pointId"), message.pointId);
out.insert(QStringLiteral("metadataId"), message.metadataId);
out.insert(QStringLiteral("timestamp"), static_cast<qint64>(message.timestamp));
out.insert(QStringLiteral("quality"), qualityToString(message.quality));
out.insert(QStringLiteral("qualityCode"), static_cast<int>(message.quality));
out.insert(QStringLiteral("value"), domValueToJson(message.value));
QJsonObject attrs;
for (auto it = message.attributes.constBegin(); it != message.attributes.constEnd(); ++it) {
attrs.insert(it.key(), it.value());
}
out.insert(QStringLiteral("attributes"), attrs);
QJsonArray trace;
for (const auto& item : message.traceLog) {
trace.push_back(item);
}
out.insert(QStringLiteral("traceLog"), trace);
return out;
}
QJsonObject enrichedMessageToJson(const softbus::core::models::EnrichedMessage& message)
{
softbus::core::models::DOMMessage dom;
dom.messageId = message.sourceMessageId;
dom.objectType = message.objectType;
dom.objectRef = message.objectRef;
dom.domainKey = message.domainKey;
dom.pointId = message.pointId;
dom.metadataId = message.metadataId;
dom.timestamp = QDateTime::currentMSecsSinceEpoch();
dom.quality = message.quality;
dom.value = message.engineeringValue;
dom.attributes = message.attributes;
dom.traceLog = message.traceLog;
auto out = domMessageToJson(dom);
out.insert(QStringLiteral("logicalName"), message.logicalName);
out.insert(QStringLiteral("unit"), message.unit);
out.insert(QStringLiteral("catalogVersion"), message.catalogVersion);
out.insert(QStringLiteral("alarm"), message.alarm);
return out;
}
QJsonObject domEventToJson(const DomEventEnvelope& event)
{
QJsonObject out;
out.insert(QStringLiteral("schemaVersion"), event.schemaVersion);
out.insert(QStringLiteral("eventId"), event.eventId);
out.insert(QStringLiteral("sourceProcess"), event.sourceProcess);
out.insert(QStringLiteral("sourceEndpoint"), event.sourceEndpoint);
out.insert(QStringLiteral("emitTs"), static_cast<qint64>(event.emitTs));
out.insert(QStringLiteral("retryCount"), event.retryCount);
out.insert(QStringLiteral("dom"), event.dom);
return out;
}
bool domEventFromJson(const QJsonObject& json, DomEventEnvelope& out)
{
if (json.isEmpty()) {
return false;
}
out.schemaVersion = json.value(QStringLiteral("schemaVersion")).toInt(1);
out.eventId = json.value(QStringLiteral("eventId")).toString();
out.sourceProcess = json.value(QStringLiteral("sourceProcess")).toString(QStringLiteral("collector-core"));
out.sourceEndpoint = json.value(QStringLiteral("sourceEndpoint")).toString();
out.emitTs = json.value(QStringLiteral("emitTs")).toVariant().toLongLong();
out.retryCount = json.value(QStringLiteral("retryCount")).toInt(0);
out.dom = json.value(QStringLiteral("dom")).toObject();
if (out.eventId.isEmpty()) {
out.eventId = QUuid::createUuid().toString(QUuid::WithoutBraces);
}
if (out.emitTs <= 0) {
out.emitTs = QDateTime::currentMSecsSinceEpoch();
}
return !out.dom.isEmpty();
}
DomEventEnvelope makeEventFromEnriched(const softbus::core::models::EnrichedMessage& message,
const QString& sourceEndpoint)
{
DomEventEnvelope event;
event.schemaVersion = 1;
event.eventId = QUuid::createUuid().toString(QUuid::WithoutBraces);
event.sourceProcess = QStringLiteral("collector-core");
event.sourceEndpoint = sourceEndpoint;
event.emitTs = QDateTime::currentMSecsSinceEpoch();
event.retryCount = 0;
event.dom = enrichedMessageToJson(message);
event.dom.insert(QStringLiteral("quality"), qualityToString(message.quality));
event.dom.insert(QStringLiteral("qualityCode"), static_cast<int>(message.quality));
return event;
}
} // namespace softbus::message_bus::ipc

34
src/message_bus/ipc/DomEvent.h Normal file
View File

@@ -0,0 +1,34 @@
#pragma once
#include <cstdint>
#include <QJsonObject>
#include <QString>
#include "core/models/DOMMessage.h"
#include "core/models/EnrichedMessage.h"
namespace softbus::message_bus::ipc
{
struct DomEventEnvelope
{
int schemaVersion{1};
QString eventId;
QString sourceProcess{QStringLiteral("collector-core")};
QString sourceEndpoint;
std::int64_t emitTs{0};
int retryCount{0};
QJsonObject dom;
};
QJsonObject domMessageToJson(const softbus::core::models::DOMMessage& message);
QJsonObject enrichedMessageToJson(const softbus::core::models::EnrichedMessage& message);
QJsonObject domEventToJson(const DomEventEnvelope& event);
bool domEventFromJson(const QJsonObject& json, DomEventEnvelope& out);
DomEventEnvelope makeEventFromEnriched(const softbus::core::models::EnrichedMessage& message,
const QString& sourceEndpoint);
} // namespace softbus::message_bus::ipc

178
src/message_bus/ipc/DomIpcPublisher.cpp Normal file
View File

@@ -0,0 +1,178 @@
#include "message_bus/ipc/DomIpcPublisher.h"
#include <QDateTime>
#include <QDir>
#include <QFile>
#include <QFileInfo>
#include <QJsonDocument>
#include <QLocalSocket>
#include <QSaveFile>
#include "message_bus/ipc/DomEvent.h"
#include "utils/logs/logging.h"
namespace softbus::message_bus::ipc
{
DomIpcPublisher& DomIpcPublisher::instance()
{
static DomIpcPublisher p;
return p;
}
void DomIpcPublisher::configure(const DomIpcConfig& cfg)
{
std::lock_guard<std::mutex> lk(m_mutex);
m_cfg = cfg;
if (m_cfg.outboxPath.isEmpty()) {
m_cfg.outboxPath = QDir::homePath() + QStringLiteral("/softbus/run/dom_outbox.ndjson");
}
}
DomIpcConfig DomIpcPublisher::config() const
{
std::lock_guard<std::mutex> lk(m_mutex);
return m_cfg;
}
void DomIpcPublisher::publishEnriched(const softbus::core::models::EnrichedMessage& message, const QString& sourceEndpoint)
{
DomIpcConfig cfg;
{
std::lock_guard<std::mutex> lk(m_mutex);
cfg = m_cfg;
}
if (!cfg.enabled) {
return;
}
DomEventEnvelope evt = makeEventFromEnriched(message, sourceEndpoint);
const QByteArray line = QJsonDocument(domEventToJson(evt)).toJson(QJsonDocument::Compact) + '\n';
if (sendJsonLine(line)) {
++m_sendOk;
replayOutbox();
return;
}
++m_sendFail;
appendOutbox(line);
}
QJsonObject DomIpcPublisher::statusJson() const
{
QJsonObject out;
out.insert(QStringLiteral("sendOk"), static_cast<qint64>(m_sendOk.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("sendFail"), static_cast<qint64>(m_sendFail.load(std::memory_order_relaxed)));
out.insert(QStringLiteral("replayed"), static_cast<qint64>(m_replayed.load(std::memory_order_relaxed)));
{
std::lock_guard<std::mutex> lk(m_mutex);
out.insert(QStringLiteral("enabled"), m_cfg.enabled);
out.insert(QStringLiteral("mirrorOnly"), m_cfg.mirrorOnly);
out.insert(QStringLiteral("socketName"), m_cfg.socketName);
out.insert(QStringLiteral("outboxPath"), m_cfg.outboxPath);
out.insert(QStringLiteral("outboxDepth"), outboxDepthNoLock());
}
return out;
}
bool DomIpcPublisher::sendJsonLine(const QByteArray& line)
{
DomIpcConfig cfg;
{
std::lock_guard<std::mutex> lk(m_mutex);
cfg = m_cfg;
}
QLocalSocket socket;
socket.connectToServer(cfg.socketName, QIODevice::WriteOnly);
if (!socket.waitForConnected(120)) {
return false;
}
if (socket.write(line) < 0) {
return false;
}
if (!socket.waitForBytesWritten(120)) {
return false;
}
socket.disconnectFromServer();
return true;
}
void DomIpcPublisher::appendOutbox(const QByteArray& line)
{
std::lock_guard<std::mutex> lk(m_mutex);
QFileInfo fi(m_cfg.outboxPath);
(void)QDir().mkpath(fi.dir().absolutePath());
QFile f(m_cfg.outboxPath);
if (!f.open(QIODevice::WriteOnly | QIODevice::Append | QIODevice::Text)) {
LOG_WARNING() << "DomIpcPublisher: failed to append outbox path=" << m_cfg.outboxPath;
return;
}
f.write(line);
f.close();
}
void DomIpcPublisher::replayOutbox()
{
DomIpcConfig cfg;
{
std::lock_guard<std::mutex> lk(m_mutex);
cfg = m_cfg;
}
if (cfg.outboxPath.isEmpty()) {
return;
}
QFile f(cfg.outboxPath);
if (!f.exists()) {
return;
}
if (!f.open(QIODevice::ReadOnly | QIODevice::Text)) {
return;
}
QList<QByteArray> lines;
while (!f.atEnd()) {
const QByteArray line = f.readLine();
if (!line.trimmed().isEmpty()) {
lines.push_back(line);
}
}
f.close();
if (lines.isEmpty()) {
(void)QFile::remove(cfg.outboxPath);
return;
}
QList<QByteArray> remain;
int replayed = 0;
for (const auto& line : lines) {
if (replayed < cfg.maxReplayPerPublish && sendJsonLine(line)) {
++replayed;
++m_replayed;
} else {
remain.push_back(line);
}
}
QSaveFile out(cfg.outboxPath);
if (!out.open(QIODevice::WriteOnly | QIODevice::Text)) {
return;
}
for (const auto& line : remain) {
out.write(line);
}
out.commit();
}
qint64 DomIpcPublisher::outboxDepthNoLock() const
{
QFile f(m_cfg.outboxPath);
if (!f.exists() || !f.open(QIODevice::ReadOnly | QIODevice::Text)) {
return 0;
}
qint64 lines = 0;
while (!f.atEnd()) {
if (!f.readLine().trimmed().isEmpty()) {
++lines;
}
}
return lines;
}
} // namespace softbus::message_bus::ipc

50
src/message_bus/ipc/DomIpcPublisher.h Normal file
View File

@@ -0,0 +1,50 @@
#pragma once
#include <atomic>
#include <mutex>
#include <QJsonObject>
#include <QString>
#include "core/models/EnrichedMessage.h"
namespace softbus::message_bus::ipc
{
struct DomIpcConfig
{
bool enabled{false};
bool mirrorOnly{true};
QString socketName{QStringLiteral("softbus_orchestrator_bus")};
QString outboxPath;
int maxReplayPerPublish{16};
};
class DomIpcPublisher
{
public:
static DomIpcPublisher& instance();
void configure(const DomIpcConfig& cfg);
DomIpcConfig config() const;
void publishEnriched(const softbus::core::models::EnrichedMessage& message, const QString& sourceEndpoint);
QJsonObject statusJson() const;
private:
DomIpcPublisher() = default;
bool sendJsonLine(const QByteArray& line);
void appendOutbox(const QByteArray& line);
void replayOutbox();
qint64 outboxDepthNoLock() const;
private:
mutable std::mutex m_mutex;
DomIpcConfig m_cfg;
std::atomic<std::uint64_t> m_sendOk{0};
std::atomic<std::uint64_t> m_sendFail{0};
std::atomic<std::uint64_t> m_replayed{0};
};
} // namespace softbus::message_bus::ipc