MQTTSuite
Loading...
Searching...
No Matches
mqttbridge.cpp
Go to the documentation of this file.
1/*
2 * MQTTSuite - A lightweight MQTT Integration System
3 * Copyright (C) Volker Christian <me@vchrist.at>
4 * 2022, 2023, 2024, 2025, 2026
5 *
6 * This program is free software: you can redistribute it and/or modify it
7 * under the terms of the GNU General Public License as published by the Free
8 * Software Foundation, either version 3 of the License, or (at your option)
9 * any later version.
10 *
11 * This program is distributed in the hope that it will be useful, but WITHOUT
12 * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or
13 * FITNESS FOR A PARTICULAR PURPOSE. See the GNU General Public License for
14 * more details.
15 *
16 * You should have received a copy of the GNU General Public License along
17 * with this program. If not, see <https://www.gnu.org/licenses/>.
18 */
19
20/*
21 * MIT License
22 *
23 * Permission is hereby granted, free of charge, to any person obtaining a copy
24 * of this software and associated documentation files (the "Software"), to deal
25 * in the Software without restriction, including without limitation the rights
26 * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
27 * copies of the Software, and to permit persons to whom the Software is
28 * furnished to do so, subject to the following conditions:
29 *
30 * The above copyright notice and this permission notice shall be included in
31 * all copies or substantial portions of the Software.
32 *
33 * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
34 * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
35 * FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
36 * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
37 * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
38 * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
39 * THE SOFTWARE.
40 */
41
42#include "ConfigBridge.h"
43#include "SocketContextFactory.h"
44#include "config.h"
45#include "lib/BridgeStore.h"
46#include "lib/Mqtt.h"
47#include "lib/SSEDistributor.h"
48
49#include <core/SNodeC.h>
50#include <express/legacy/in/Server.h>
51#include <express/middleware/JsonMiddleware.h>
52#include <express/middleware/StaticMiddleware.h>
53#include <express/tls/in/Server.h>
54#include <iot/mqtt/MqttContext.h>
55#include <utils/Config.h>
56//
57
58#ifndef DOXYGEN_SHOULD_SKIP_THIS
59
60//
61#include <log/Logger.h>
62//
63#include <nlohmann/json_fwd.hpp>
64//
65
66// Select necessary include files
67// ==============================
68#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4)
69#include <net/in/stream/legacy/SocketClient.h>
70#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4)
71#include <net/in/stream/tls/SocketClient.h>
72#endif
73#endif
74
75#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6)
76#include <net/in6/stream/legacy/SocketClient.h>
77#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6)
78#include <net/in6/stream/tls/SocketClient.h>
79#endif
80#endif
81
82#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX)
83#include <net/un/stream/legacy/SocketClient.h>
84#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS)
85#include <net/un/stream/tls/SocketClient.h>
86#endif
87#endif
88
89#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
90#include <web/http/legacy/in/Client.h>
91#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
92#include <web/http/tls/in/Client.h>
93#endif
94#endif
95
96#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
97#include <web/http/legacy/in6/Client.h>
98#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
99#include <web/http/tls/in6/Client.h>
100#endif
101#endif
102
103#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
104#include <web/http/legacy/un/Client.h>
105#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
106#include <web/http/tls/un/Client.h>
107#endif
108#endif
109
110#include <list>
111#include <nlohmann/json.hpp>
112
113#endif
114
115static std::map<uint64_t, core::socket::stream::ClientFlowController*> flowControllers;
116
117static bool restart = false;
118
119static void startBridges();
120
121static void restartBridges() {
122 if (restart) {
123 VLOG(2) << "Restarting bridges...";
124
125 mqtt::bridge::lib::BridgeStore::instance().activateStaged();
126
127 startBridges();
128
129 utils::Config::parse();
130
131 restart = false;
132 } else {
133 VLOG(2) << "No bridge restarted";
134 }
135}
136
137static void handleFlowControllers(core::socket::stream::ClientFlowController* clientFlowController) {
138 flowControllers.emplace(clientFlowController->getId(), clientFlowController);
139 VLOG(2) << "Added FlowController for: [" << clientFlowController->getId() << "] " << clientFlowController->getInstanceName();
140
141 clientFlowController->setOnFlowCompleted([](uint64_t id, const std::string& instanceName) {
142 flowControllers.erase(id);
143 VLOG(2) << "Erased FlowController of: [" << id << "] " << instanceName;
144
145 if (flowControllers.empty() && restart) {
146 restartBridges();
147 }
148 });
149}
150
151static bool closeBridges() {
152 restart = true;
153
154 if (!flowControllers.empty()) {
155 mqtt::bridge::lib::SSEDistributor::instance().bridgesStopping();
156
157 for (const auto& [bridgeName, bridge] : mqtt::bridge::lib::BridgeStore::instance().getBridgeMap()) {
158 mqtt::bridge::lib::SSEDistributor::instance().bridgeStopping(bridgeName);
159
160 for (const auto& mqtt : bridge.getMqttList()) {
161 mqtt::bridge::lib::SSEDistributor::instance().brokerDisconnecting(
162 bridgeName, mqtt->getMqttContext()->getSocketConnection()->getInstanceName());
163
164 mqtt->sendDisconnect();
165 }
166 }
167
168 for (auto& [id, flowController] : flowControllers) {
169 VLOG(1) << "Terminating Flow of: [" << flowController->getId() << "] " << flowController->getInstanceName();
170
171 flowController->terminateFlow();
172 }
173 }
174
175 return flowControllers.empty();
176}
177
178static void
179reportState(const std::string& instanceName, const core::socket::SocketAddress& socketAddress, const core::socket::State& state) {
180 switch (state) {
181 case core::socket::State::OK:
182 VLOG(1) << instanceName << ": connected to '" << socketAddress.toString() << "'";
183 break;
184 case core::socket::State::DISABLED:
185 VLOG(1) << instanceName << ": disabled";
186 break;
187 case core::socket::State::ERROR:
188 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
189 break;
190 case core::socket::State::FATAL:
191 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
192 break;
193 }
194}
195
196template <template <typename SocketContextFactoryT, typename... ArgsT> typename SocketClient>
197static SocketClient<mqtt::bridge::SocketContextFactory> startClient( //
198 const std::string& instanceName,
199 const std::function<void(typename SocketClient<mqtt::bridge::SocketContextFactory>::Config*)>& configurator) {
200 using Client = SocketClient<mqtt::bridge::SocketContextFactory>;
201 using SocketAddress = typename Client::SocketAddress;
202
203 Client socketClient = core::socket::stream::Client<Client>(instanceName, configurator);
204
205 socketClient.getConfig()->Instance::configurable(false);
206 socketClient.getConfig()->Remote::configurable(false);
207 socketClient.getConfig()->setRetry()->setRetryBase(1);
208 socketClient.getConfig()->setReconnect();
209
210 handleFlowControllers(socketClient.getFlowController());
211
212 socketClient.connect([instanceName](const SocketAddress& socketAddress, const core::socket::State& state) {
213 reportState(instanceName, socketAddress, state);
214 });
215
216 return socketClient;
217}
218
219template <typename HttpClient>
220static HttpClient startClient( //
221 const std::string& instanceName,
222 const std::function<void(typename HttpClient::Config*)>& configurator) {
223 using SocketAddress = typename HttpClient::SocketAddress;
224
225 HttpClient httpClient(
226 instanceName,
227 [](const std::shared_ptr<web::http::client::MasterRequest>& req) {
228 const std::string connectionName = req->getSocketContext()->getSocketConnection()->getConnectionName();
229
230 req->set("Sec-WebSocket-Protocol", "mqtt");
231
232 req->upgrade(
233 "/ws",
234 "websocket",
235 [connectionName](bool success) {
236 VLOG(1) << connectionName << ": HTTP Upgrade (http -> websocket||"
237 << "mqtt" << ") start " << (success ? "success" : "failed");
238 },
239 []([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req,
240 [[maybe_unused]] const std::shared_ptr<web::http::client::Response>& res,
241 [[maybe_unused]] bool success) {
242 },
243 [connectionName]([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req, const std::string& message) {
244 VLOG(1) << connectionName << ": Request parse error: " << message;
245 });
246 },
247 []([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req) {
248 VLOG(1) << "Session ended";
249 });
250
251 configurator(httpClient.getConfig());
252
253 httpClient.getConfig()->Instance::configurable(false);
254 httpClient.getConfig()->Remote::configurable(false);
255 httpClient.getConfig()->setRetry()->setRetryBase(1);
256 httpClient.getConfig()->setReconnect();
257
258 handleFlowControllers(httpClient.getFlowController());
259
260 httpClient.connect([instanceName](const SocketAddress& socketAddress, const core::socket::State& state) {
261 reportState(instanceName, socketAddress, state);
262 });
263
264 return httpClient;
265}
266
267static void startBridges() {
268 mqtt::bridge::lib::SSEDistributor::instance().bridgesStarting();
269
270 for (const auto& [bridgeName, bridge] : mqtt::bridge::lib::BridgeStore::instance().getBridgeMap()) {
271 VLOG(0) << "Starting bridge: " << bridgeName;
272
273 if (!bridge.getDisabled()) {
274 mqtt::bridge::lib::SSEDistributor::instance().bridgeStarting(bridgeName);
275
276 for (const auto& [fullInstanceName, broker] : bridge.getBrokerMap()) {
277 if (!broker.getDisabled()) {
278 mqtt::bridge::lib::SSEDistributor::instance().brokerConnecting(bridgeName, fullInstanceName);
279
280 VLOG(1) << " Creating broker instance: " << fullInstanceName;
281 VLOG(1) << " Broker prefix: " << broker.getPrefix();
282 VLOG(1) << " Broker client id: " << broker.getClientId();
283 VLOG(1) << " Broker disabled: " << broker.getDisabled();
284 VLOG(1) << " Broker address: " << broker.getAddress();
285 VLOG(1) << " Broker prefix: " << broker.getPrefix();
286 VLOG(1) << " Broker username: " << broker.getUsername();
287 VLOG(1) << " Broker password: " << broker.getPassword();
288 VLOG(1) << " Broker client-id: " << broker.getClientId();
289 VLOG(1) << " Broker clean session: " << broker.getCleanSession();
290 VLOG(1) << " Broker will-topic: " << broker.getWillTopic();
291 VLOG(1) << " Broker will-message: " << broker.getWillMessage();
292 VLOG(1) << " Broker will-qos: " << static_cast<int>(broker.getWillQoS());
293 VLOG(1) << " Broker will-retain: " << broker.getWillRetain();
294 VLOG(1) << " Broker loop prevention: " << broker.getLoopPrevention();
295 VLOG(1) << " Bridge disabled: " << bridge.getDisabled();
296 VLOG(1) << " Bridge prefix: " << bridge.getPrefix();
297 VLOG(1) << " Bridge Transport: " << broker.getTransport();
298 VLOG(1) << " Bridge Protocol: " << broker.getProtocol();
299 VLOG(1) << " Bridge Encryption: " << broker.getEncryption();
300
301 VLOG(1) << " Topics:";
302 const std::list<iot::mqtt::Topic>& topics = broker.getTopics();
303 for (const iot::mqtt::Topic& topic : topics) {
304 VLOG(1) << " " << topic.getName() << ":" << static_cast<uint16_t>(topic.getQoS());
305 }
306
307 const std::string& transport = broker.getTransport();
308 const std::string& protocol = broker.getProtocol();
309 const std::string& encryption = broker.getEncryption();
310
311 if (transport == "stream") {
312 if (protocol == "in") {
313 if (encryption == "legacy") {
314#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4)
315 startClient<net::in::stream::legacy::SocketClient>( //
316 fullInstanceName,
317 [&broker](net::in::stream::legacy::config::ConfigSocketClient* config) {
318 config->setDisableNagleAlgorithm();
319
320 config->Remote::setHost(broker.getAddress()["host"]);
321 config->Remote::setPort(broker.getAddress()["port"]);
322
323 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
324 });
325#else // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4
326 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
327 << "' not supported.";
328#endif // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4
329 } else if (encryption == "tls") {
330#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4)
331 startClient<net::in::stream::tls::SocketClient>( //
332 fullInstanceName,
333 [&broker](net::in::stream::tls::config::ConfigSocketClient* config) {
334 config->setDisableNagleAlgorithm();
335
336 config->Remote::setHost(broker.getAddress()["host"]);
337 config->Remote::setPort(broker.getAddress()["port"]);
338
339 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
340 });
341#else // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4
342 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
343 << "' not supported.";
344#endif // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4
345 }
346 } else if (protocol == "in6") {
347 if (encryption == "legacy") {
348#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6)
349 startClient<net::in6::stream::legacy::SocketClient>( //
350 fullInstanceName,
351 [&broker](net::in6::stream::legacy::config::ConfigSocketClient* config) {
352 config->setDisableNagleAlgorithm();
353
354 config->Remote::setHost(broker.getAddress()["host"]);
355 config->Remote::setPort(broker.getAddress()["port"]);
356
357 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
358 });
359#else // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6
360 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
361 << "' not supported.";
362#endif // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6
363 } else if (encryption == "tls") {
364#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6)
365 startClient<net::in6::stream::tls::SocketClient>( //
366 fullInstanceName,
367 [&broker](net::in6::stream::tls::config::ConfigSocketClient* config) {
368 config->setDisableNagleAlgorithm();
369
370 config->Remote::setHost(broker.getAddress()["host"]);
371 config->Remote::setPort(broker.getAddress()["port"]);
372
373 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
374 });
375#else // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6
376 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
377 << "' not supported.";
378#endif // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6
379 }
380 } else if (protocol == "un") {
381 if (encryption == "legacy") {
382#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX)
383 startClient<net::un::stream::legacy::SocketClient>( //
384 fullInstanceName,
385 [&broker](net::un::stream::legacy::config::ConfigSocketClient* config) {
386 config->Remote::setSunPath(broker.getAddress()["host"]);
387
388 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
389 });
390#else // CONFIG_MQTTSUITE_BRIDGE_UNIX
391 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
392 << "' not supported.";
393#endif // CONFIG_MQTTSUITE_BRIDGE_UNIX
394 } else if (encryption == "tls") {
395#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS)
396 startClient<net::un::stream::tls::SocketClient>( //
397 fullInstanceName,
398 [&broker](net::un::stream::tls::config::ConfigSocketClient* config) {
399 config->Remote::setSunPath(broker.getAddress()["host"]);
400
401 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
402 });
403#else // CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS
404 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
405 << "' not supported.";
406#endif // CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS
407 }
408 }
409 } else if (transport == "websocket") {
410 if (protocol == "in") {
411 if (encryption == "legacy") {
412#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
413 startClient<web::http::legacy::in::Client>( //
414 fullInstanceName,
415 [&broker](net::in::stream::legacy::config::ConfigSocketClient* config) {
416 config->setDisableNagleAlgorithm();
417
418 config->Remote::setHost(broker.getAddress()["host"]);
419 config->Remote::setPort(broker.getAddress()["port"]);
420
421 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
422 });
423#else // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4 && CONFIG_MQTTSUITE_BRIDGE_WS
424 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
425 << "' not supported.";
426#endif // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV4 && CONFIG_MQTTSUITE_BRIDGE_WS
427 } else if (encryption == "tls") {
428#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
429 startClient<web::http::tls::in::Client>( //
430 fullInstanceName,
431 [&broker](net::in::stream::tls::config::ConfigSocketClient* config) {
432 config->setDisableNagleAlgorithm();
433
434 config->Remote::setHost(broker.getAddress()["host"]);
435 config->Remote::setPort(broker.getAddress()["port"]);
436
437 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
438 });
439#else // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4 && CONFIG_MQTTSUITE_BRIDGE_WSS
440 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
441 << "' not supported.";
442#endif // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV4 && CONFIG_MQTTSUITE_BRIDGE_WSS
443 }
444 } else if (protocol == "in6") {
445 if (encryption == "legacy") {
446#if defined(CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
447 startClient<web::http::legacy::in6::Client>( //
448 fullInstanceName,
449 [&broker](net::in6::stream::legacy::config::ConfigSocketClient* config) {
450 config->setDisableNagleAlgorithm();
451
452 config->Remote::setHost(broker.getAddress()["host"]);
453 config->Remote::setPort(broker.getAddress()["port"]);
454
455 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
456 });
457#else // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6 && CONFIG_MQTTSUITE_BRIDGE_WS
458 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
459 << "' not supported.";
460#endif // CONFIG_MQTTSUITE_BRIDGE_TCP_IPV6&& CONFIG_MQTTSUITE_BRIDGE_WS
461 } else if (encryption == "tls") {
462#if defined(CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
463 startClient<web::http::tls::in6::Client>( //
464 fullInstanceName,
465 [&broker](net::in6::stream::tls::config::ConfigSocketClient* config) {
466 config->setDisableNagleAlgorithm();
467
468 config->Remote::setHost(broker.getAddress()["host"]);
469 config->Remote::setPort(broker.getAddress()["port"]);
470
471 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
472 });
473#else // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6 && CONFIG_MQTTSUITE_BRIDGE_WSS
474 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
475 << "' not supported.";
476#endif // CONFIG_MQTTSUITE_BRIDGE_TLS_IPV6 && CONFIG_MQTTSUITE_BRIDGE_WSS
477 }
478 } else if (protocol == "un") {
479 if (encryption == "legacy") {
480#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX) && defined(CONFIG_MQTTSUITE_BRIDGE_WS)
481 startClient<web::http::legacy::un::Client>( //
482 fullInstanceName,
483 [&broker](net::un::stream::legacy::config::ConfigSocketClient* config) {
484 config->Remote::setSunPath(broker.getAddress()["path"]);
485
486 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
487 });
488#else // CONFIG_MQTTSUITE_BRIDGE_UNIX && CONFIG_MQTTSUITE_BRIDGE_WS
489 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
490 << "' not supported.";
491#endif // CONFIG_MQTTSUITE_BRIDGE_UNIX && CONFIG_MQTTSUITE_BRIDGE_WS
492 } else if (encryption == "tls") {
493#if defined(CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS) && defined(CONFIG_MQTTSUITE_BRIDGE_WSS)
494 startClient<web::http::tls::un::Client>( //
495 fullInstanceName,
496 [&broker](net::un::stream::tls::config::ConfigSocketClient* config) {
497 config->Remote::setSunPath(broker.getAddress()["path"]);
498
499 config->setDisabled(broker.getDisabled() || broker.getBridge().getDisabled());
500 });
501#else // CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS && CONFIG_MQTTSUITE_BRIDGE_WSS
502 VLOG(1) << " Transport '" << transport << "', protocol '" << protocol << "', encryption '" << encryption
503 << "' not supported.";
504#endif // CONFIG_MQTTSUITE_BRIDGE_UNIX_TLS && CONFIG_MQTTSUITE_BRIDGE_WSS
505 }
506 }
507 } else {
508 VLOG(1) << " Transport '" << transport << "' not supported.";
509 }
510 } else {
511 mqtt::bridge::lib::SSEDistributor::instance().brokerDisabled(bridgeName, fullInstanceName);
512 }
513 }
514 } else {
515 mqtt::bridge::lib::SSEDistributor::instance().bridgeDisabled(bridgeName);
516 }
517 }
518}
519
520int main(int argc, char* argv[]) {
521 utils::Config::configRoot.newSubCommand<mqtt::bridge::ConfigBridge>();
522
523 core::SNodeC::init(argc, argv);
524
525 const express::Router router(express::middleware::JsonMiddleware());
526
527 router.get("/api/bridge/config", [] APPLICATION(req, res) { // cppcheck-suppress unknownMacro
528 res->send(mqtt::bridge::lib::BridgeStore::instance().getBridgesConfigJson().dump(4));
529 });
530
531 router.patch("/api/bridge/config", [] APPLICATION(req, res) {
532 req->getAttribute<nlohmann::json>(
533 [&res](nlohmann::json& jsonPatch) {
534 if (!restart) {
535 if (mqtt::bridge::lib::BridgeStore::instance().patch(jsonPatch)) {
536 res->send(R"({"success": true, "message": "Bridge config patch applied"})"_json.dump());
537
538 if (closeBridges()) {
539 restartBridges();
540 }
541 } else {
542 res->status(404).send(R"({"success": false, "message": "Bridge config patch failed to applie"})"_json.dump());
543 }
544 } else {
545 res->status(409).send(
546 R"({"success": false, "message": "Bridge is in restarting state. Patch not applied"})"_json.dump());
547 }
548 },
549 [&res](const std::string& key) {
550 VLOG(1) << "Attribute type not found: " << key;
551
552 res->status(400).send("Attribute type not found: " + key);
553 });
554 });
555
556 router.get("/api/bridge/sse", [] APPLICATION(req, res) {
557 if (web::http::ciContains(req->get("Accept"), "text/event-stream")) {
558 res->set("Content-Type", "text/event-stream") //
559 .set("Cache-Control", "no-cache")
560 .set("Connection", "keep-alive");
561 res->sendHeader();
562
563 std::string data{"data"};
564 mqtt::bridge::lib::SSEDistributor::instance().addEventReceiver(res, req->get("Last-Event-ID"));
565 } else {
566 res->redirect("/clients");
567 }
568 });
569
570 router.get("/", [] APPLICATION(req, res) {
571 res->redirect("/config");
572 });
573
574 router.get("/config", [] APPLICATION(req, res) {
575 res->redirect("/config/index.html");
576 });
577
578 router.use("/config",
579 express::middleware::StaticMiddleware(utils::Config::configRoot.getSubCommand<mqtt::bridge::ConfigBridge>()->getHtmlDir()));
580
581 router.get("*", [] APPLICATION(req, res) {
582 res->redirect("/config/index.html");
583 });
584
585 express::legacy::in::Server( //
586 "admin-legacy",
587 router,
588 reportState, //
589 [](net::in::stream::legacy::config::ConfigSocketServer* config) {
590 config->setPort(8081);
591 config->setRetry();
592 config->setReuseAddress();
593 });
594
595 express::tls::in::Server( //
596 "admin-tls",
597 router,
598 reportState,
599 [](net::in::stream::tls::config::ConfigSocketServer* config) {
600 config->setPort(8082);
601 config->setRetry();
602 config->setReuseAddress();
603 });
604
605 if (mqtt::bridge::lib::BridgeStore::instance().loadAndValidate(
606 utils::Config::configRoot.getSubCommand<mqtt::bridge::ConfigBridge>()->getDefinitionFile())) {
607 startBridges();
608 } else {
609 VLOG(1) << "Loading bridge definition file failed";
610 }
611
612 return core::SNodeC::start();
613}
mqtt::bridge::lib::BridgeStore::instance
static BridgeStore & instance()
Definition BridgeStore.cpp:65
mqtt::bridge::lib::BridgeStore::loadAndValidate
bool loadAndValidate(const std::string &fileName)
Definition BridgeStore.cpp:71
mqtt::bridge::lib::SSEDistributor::addEventReceiver
void addEventReceiver(const std::shared_ptr< express::Response > &response, const std::string &lastEventId)
Definition SSEDistributor.cpp:74
mqtt::bridge::lib::SSEDistributor::instance
static SSEDistributor & instance()
Definition SSEDistributor.cpp:68
main
int main(int argc, char *argv[])
Definition json-schema-validate.cpp:46
startClient
static SocketClient< mqtt::bridge::SocketContextFactory > startClient(const std::string &instanceName, const std::function< void(typename SocketClient< mqtt::bridge::SocketContextFactory >::Config *)> &configurator)
Definition mqttbridge.cpp:197
closeBridges
static bool closeBridges()
Definition mqttbridge.cpp:151
flowControllers
static std::map< uint64_t, core::socket::stream::ClientFlowController * > flowControllers
Definition mqttbridge.cpp:115
handleFlowControllers
static void handleFlowControllers(core::socket::stream::ClientFlowController *clientFlowController)
Definition mqttbridge.cpp:137
reportState
static void reportState(const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
Definition mqttbridge.cpp:179
startBridges
static void startBridges()
Definition mqttbridge.cpp:267
restart
static bool restart
Definition mqttbridge.cpp:117
restartBridges
static void restartBridges()
Definition mqttbridge.cpp:121
startClient
static HttpClient startClient(const std::string &instanceName, const std::function< void(typename HttpClient::Config *)> &configurator)
Definition mqttbridge.cpp:220