MQTTSuite
Loading...
Searching...
No Matches
mqttbridge.cpp File Reference
#include "ConfigBridge.h"
#include "SocketContextFactory.h"
#include "config.h"
#include "lib/BridgeStore.h"
#include "lib/Mqtt.h"
#include "lib/SSEDistributor.h"
#include <core/SNodeC.h>
#include <express/legacy/in/Server.h>
#include <express/middleware/JsonMiddleware.h>
#include <express/middleware/StaticMiddleware.h>
#include <express/tls/in/Server.h>
#include <iot/mqtt/MqttContext.h>
#include <utils/Config.h>
Include dependency graph for mqttbridge.cpp:

Go to the source code of this file.

Functions

static void startBridges ()
static void restartBridges ()
static void handleFlowControllers (core::socket::stream::ClientFlowController *clientFlowController)
static bool closeBridges ()
static void reportState (const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
template<template< typename SocketContextFactoryT, typename... ArgsT > typename SocketClient>
static SocketClient< mqtt::bridge::SocketContextFactory > startClient (const std::string &instanceName, const std::function< void(typename SocketClient< mqtt::bridge::SocketContextFactory >::Config *)> &configurator)
template<typename HttpClient>
static HttpClient startClient (const std::string &instanceName, const std::function< void(typename HttpClient::Config *)> &configurator)
int main (int argc, char *argv[])

Variables

static std::map< uint64_t, core::socket::stream::ClientFlowController * > flowControllers
static bool restart = false

Function Documentation

◆ closeBridges()

bool closeBridges ( )
static

Definition at line 151 of file mqttbridge.cpp.

151 {
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}
mqtt::bridge::lib::BridgeStore::instance
static BridgeStore & instance()
Definition BridgeStore.cpp:65
mqtt::bridge::lib::SSEDistributor::instance
static SSEDistributor & instance()
Definition SSEDistributor.cpp:68
mqtt::bridge::lib::SSEDistributor::brokerDisconnecting
void brokerDisconnecting(const std::string &bridgeName, const std::string &instanceName)
Definition SSEDistributor.cpp:191
mqtt::bridge::lib::SSEDistributor::bridgeStopping
void bridgeStopping(const std::string &bridgeName)
Definition SSEDistributor.cpp:163
flowControllers
static std::map< uint64_t, core::socket::stream::ClientFlowController * > flowControllers
Definition mqttbridge.cpp:115
restart
static bool restart
Definition mqttbridge.cpp:117

References mqtt::bridge::lib::SSEDistributor::bridgesStopping(), mqtt::bridge::lib::SSEDistributor::instance(), and restart.

Here is the call graph for this function:

◆ handleFlowControllers()

void handleFlowControllers ( core::socket::stream::ClientFlowController * clientFlowController)
static

Definition at line 137 of file mqttbridge.cpp.

137 {
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}
restartBridges
static void restartBridges()
Definition mqttbridge.cpp:121

References flowControllers, restart, and restartBridges().

Here is the call graph for this function:

◆ main()

int main ( int argc,
char * argv[] )

Definition at line 520 of file mqttbridge.cpp.

520 {
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::SSEDistributor::addEventReceiver
void addEventReceiver(const std::shared_ptr< express::Response > &response, const std::string &lastEventId)
Definition SSEDistributor.cpp:74
closeBridges
static bool closeBridges()
Definition mqttbridge.cpp:151
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

References mqtt::bridge::lib::SSEDistributor::addEventReceiver(), mqtt::bridge::lib::BridgeStore::instance(), mqtt::bridge::lib::SSEDistributor::instance(), mqtt::bridge::lib::BridgeStore::loadAndValidate(), reportState(), and startBridges().

Here is the call graph for this function:

◆ reportState()

void reportState ( const std::string & instanceName,
const core::socket::SocketAddress & socketAddress,
const core::socket::State & state )
static

Definition at line 179 of file mqttbridge.cpp.

179 {
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}

Referenced by main().

Here is the caller graph for this function:

◆ restartBridges()

void restartBridges ( )
static

Definition at line 121 of file mqttbridge.cpp.

121 {
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}

References mqtt::bridge::lib::BridgeStore::activateStaged(), mqtt::bridge::lib::BridgeStore::instance(), restart, and startBridges().

Referenced by handleFlowControllers().

Here is the call graph for this function:
Here is the caller graph for this function:

◆ startBridges()

void startBridges ( )
static

Definition at line 267 of file mqttbridge.cpp.

267 {
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}
mqtt::bridge::lib::SSEDistributor::brokerConnecting
void brokerConnecting(const std::string &bridgeName, const std::string &instanceName)
Definition SSEDistributor.cpp:179
mqtt::bridge::lib::SSEDistributor::brokerDisabled
void brokerDisabled(const std::string &bridgeName, const std::string &instanceName)
Definition SSEDistributor.cpp:173
mqtt::bridge::lib::SSEDistributor::bridgeDisabled
void bridgeDisabled(const std::string &bridgeName)
Definition SSEDistributor.cpp:148
mqtt::bridge::lib::SSEDistributor::bridgeStarting
void bridgeStarting(const std::string &bridgeName)
Definition SSEDistributor.cpp:153
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

References mqtt::bridge::lib::SSEDistributor::bridgesStarting(), and mqtt::bridge::lib::SSEDistributor::instance().

Referenced by main(), and restartBridges().

Here is the call graph for this function:
Here is the caller graph for this function:

◆ startClient() [1/2]

template<typename HttpClient>
HttpClient startClient ( const std::string & instanceName,
const std::function< void(typename HttpClient::Config *)> & configurator )
static

Definition at line 220 of file mqttbridge.cpp.

222 {
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}
handleFlowControllers
static void handleFlowControllers(core::socket::stream::ClientFlowController *clientFlowController)
Definition mqttbridge.cpp:137

◆ startClient() [2/2]

template<template< typename SocketContextFactoryT, typename... ArgsT > typename SocketClient>
SocketClient< mqtt::bridge::SocketContextFactory > startClient ( const std::string & instanceName,
const std::function< void(typename SocketClient< mqtt::bridge::SocketContextFactory >::Config *)> & configurator )
static

Definition at line 197 of file mqttbridge.cpp.

199 {
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}

Variable Documentation

◆ flowControllers

std::map< uint64_t, core::socket::stream::ClientFlowController * > flowControllers
static

Definition at line 115 of file mqttbridge.cpp.

Referenced by handleFlowControllers().

◆ restart

bool restart = false
static

Definition at line 117 of file mqttbridge.cpp.

Referenced by closeBridges(), handleFlowControllers(), and restartBridges().