init

src/message_bus/ingress/EdgeTcpUplinkPort.cpp

@@ -0,0 +1,53 @@
+#include "message_bus/ingress/EdgeTcpUplinkPort.h"
+
+#include <cstring>
+
+#include <QByteArray>
+#include <QDebug>
+
+#include "message_bus/ingress/EdgeUplinkWire.h"
+
+namespace softbus::message_bus::ingress
+{
+
+EdgeTcpUplinkPort::EdgeTcpUplinkPort(FrameWriter writer)
+    : m_writer(std::move(writer))
+{
+}
+
+bool EdgeTcpUplinkPort::push(softbus::core::models::RawBusMessage ctx)
+{
+    if (!m_writer || !ctx.payload.valid()) {
+        return false;
+    }
+    std::uint64_t protocolHeaderRaw = 0;
+    std::memcpy(&protocolHeaderRaw, &ctx.header, sizeof(protocolHeaderRaw));
+
+    const auto* bytes = ctx.payload.bytes();
+    const QByteArray busPayload(reinterpret_cast<const char*>(bytes), static_cast<int>(ctx.payload.length));
+
+    const QByteArray inner = edge_uplink::encodeDataRawBusPayload(ctx.deviceId,
+                                                                  ctx.endpointHash,
+                                                                  static_cast<std::uint16_t>(ctx.protocol),
+                                                                  ctx.logicalAddress,
+                                                                  ctx.routingKey,
+                                                                  protocolHeaderRaw,
+                                                                  ctx.timestampMs,
+                                                                  static_cast<std::int8_t>(ctx.direction),
+                                                                  ctx.traceId.isEmpty() ? QStringLiteral("edge")
+                                                                                          : ctx.traceId,
+                                                                  busPayload);
+    if (inner.isEmpty()) {
+        return false;
+    }
+    const QByteArray outer =
+        edge_uplink::encodeOuterFrame(edge_uplink::FrameKind::DataRawBus, inner);
+    return m_writer(outer);
+}
+
+void EdgeTcpUplinkPort::reportError(const QString& endpoint, const QString& error)
+{
+    qWarning() << "EdgeTcpUplinkPort" << endpoint << error;
+}
+
+} // namespace softbus::message_bus::ingress