MQTTSuite
Loading...
Searching...
No Matches
mqttintegrator.cpp File Reference
#include "SocketContextFactory.h"
#include "config.h"
#include "lib/ConfigApplication.h"
#include <core/SNodeC.h>
#include <utils/Config.h>
#include <net/in/stream/legacy/SocketClient.h>
#include <net/in/stream/tls/SocketClient.h>
#include <net/in6/stream/legacy/SocketClient.h>
#include <net/in6/stream/tls/SocketClient.h>
#include <net/un/stream/legacy/SocketClient.h>
#include <net/un/stream/tls/SocketClient.h>
#include <web/http/legacy/in/Client.h>
#include <web/http/legacy/in6/Client.h>
#include <web/http/legacy/un/Client.h>
#include <web/http/tls/in/Client.h>
#include <web/http/tls/in6/Client.h>
#include <web/http/tls/un/Client.h>
#include <express/legacy/in/Server.h>
#include <express/tls/in/Server.h>
#include "lib/MappingAdminRouter.h"
#include "lib/Mqtt.h"
Include dependency graph for mqttintegrator.cpp:

Go to the source code of this file.

Functions

static void reportState (const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
template<template< typename SocketContextFactoryT, typename... ArgsT > typename SocketClientT, typename... Args>
static SocketClientT< mqtt::mqttintegrator::SocketContextFactory, Args... > startClient (const std::string &instanceName, const std::function< void(typename SocketClientT< mqtt::mqttintegrator::SocketContextFactory >::Config *)> &configurator, Args &&... args)
template<typename HttpClient>
HttpClient startClient (const std::string &name, const std::function< void(typename HttpClient::Config *)> &configurator=nullptr)
int main (int argc, char *argv[])

Function Documentation

◆ main()

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

Definition at line 178 of file mqttintegrator.cpp.

178 {
179 mqtt::lib::ConfigMqttIntegrator* configMqttIntegrator = utils::Config::configRoot.newSubCommand<mqtt::lib::ConfigMqttIntegrator>();
180
181 configMqttIntegrator->setMappingFile("mapping.json"); // Load mapping from mapping.json. (can throw)
182
183 configMqttIntegrator->setMapping( // Override mapping from mapping.json with a in-code mapping (can throw)
184 R"(
185 {
186 "discover_prefix": "",
187 "connection": {
188 "keep_alive": 60,
189 "client_id": "",
190 "clean_session": true,
191 "will_topic": "",
192 "will_message": "",
193 "will_qos": 0,
194 "will_retain": false
195 },
196 "mapping": {
197 "plugins": [],
198 "topic_level": [{
199 "name": "value",
200 "subscription": {
201 "value": {
202 "mapped_topic": "mapping/json",
203 "mapping_template": "{\"state\":\"{{message}}\"}"
204 }
205 }
206 }]
207 }
208 }
209 )");
210 /*
211 if (configMqttIntegrator->persistMapping()) {
212 VLOG(0) << "Mapping persisted successfully";
213 } else {
214 VLOG(0) << "Mapping deploy acknowledged but not persisted";
215 }
216 */
217
218 core::SNodeC::init(argc, argv);
219
220 // Instanciate Admin Router for Mapping Management
221 express::Router router =
222 mqtt::lib::admin::makeMappingAdminRouter(configMqttIntegrator, mqtt::lib::admin::AdminOptions{}, [](bool mustReconnect) {
223 return mqtt::mqttintegrator::lib::Mqtt::updateSubscriptions(mustReconnect);
224 });
225
226 express::legacy::in::Server("in-http", router, reportState, [](net::in::stream::legacy::config::ConfigSocketServer* config) {
227 config->setPort(8085);
228 config->setRetry();
229 });
230
231 express::tls::in::Server("in-https", router, reportState, [](net::in::stream::tls::config::ConfigSocketServer* config) {
232 config->setPort(8086);
233 config->setRetry();
234 });
235
236#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TCP_IPV4)
237 startClient<net::in::stream::legacy::SocketClient>( //
238 "in-mqtt",
239 [](net::in::stream::legacy::config::ConfigSocketClient* config) {
240 config->Remote::setPort(1883);
241
242 config->setDisableNagleAlgorithm();
243 });
244#endif // CONFIG_MQTTSUITE_INTEGRATOR_TCP_IPV4
245
246#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TLS_IPV4)
247 startClient<net::in::stream::tls::SocketClient>( //
248 "in-mqtts",
249 [](net::in::stream::tls::config::ConfigSocketClient* config) {
250 config->Remote::setPort(1883);
251 config->setDisableNagleAlgorithm();
252 });
253#endif
254
255#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TCP_IPV6)
256 startClient<net::in6::stream::legacy::SocketClient>( //
257 "in6-mqtt",
258 [](net::in6::stream::legacy::config::ConfigSocketClient* config) {
259 config->Remote::setPort(1883);
260 config->setDisableNagleAlgorithm();
261 });
262#endif
263
264#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TLS_IPV6)
265 startClient<net::in6::stream::tls::SocketClient>( //
266 "in6-mqtts",
267 [](net::in6::stream::tls::config::ConfigSocketClient* config) {
268 config->Remote::setPort(1883);
269 config->setDisableNagleAlgorithm();
270 });
271#endif
272
273#if defined(CONFIG_MQTTSUITE_INTEGRATOR_UNIX)
274 startClient<net::un::stream::legacy::SocketClient>( //
275 "un-mqtt",
276 []([[maybe_unused]] const net::un::stream::legacy::config::ConfigSocketClient* config) {
277 });
278#endif
279
280#if defined(CONFIG_MQTTSUITE_INTEGRATOR_UNIX_TLS)
281 startClient<net::un::stream::tls::SocketClient>( //
282 "un-mqtts",
283 []([[maybe_unused]] const net::un::stream::tls::config::ConfigSocketClient* config) {
284 });
285#endif
286
287#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TCP_IPV4) && defined(CONFIG_MQTTSUITE_INTEGRATOR_WS)
288 startClient<web::http::legacy::in::Client>( //
289 "in-wsmqtt",
290 [](net::in::stream::legacy::config::ConfigSocketClient* config) {
291 config->Remote::setPort(8080);
292 config->setDisableNagleAlgorithm();
293 });
294#endif
295
296#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TLS_IPV4) && defined(CONFIG_MQTTSUITE_INTEGRATOR_WSS)
297 startClient<web::http::tls::in::Client>( //
298 "in-wsmqtts",
299 [](net::in::stream::tls::config::ConfigSocketClient* config) {
300 config->Remote::setPort(8088);
301 config->setDisableNagleAlgorithm();
302 });
303#endif
304
305#if defined(CONFIG_MQTTSUITE_INTEGRATOR_TCP_IPV6) && defined(CONFIG_MQTTSUITE_INTEGRATOR_WS)
306 startClient<web::http::legacy::in6::Client>( //
307 "in6-wsmqtt",
308 [](net::in6::stream::legacy::config::ConfigSocketClient* config) {
309 config->Remote::setPort(8080);
310 config->setDisableNagleAlgorithm();
311 });
312#endif
313
mqtt::lib::ConfigApplication::setMapping
bool setMapping(const std::string &mapping) const
Definition ConfigApplication.cpp:119
mqtt::lib::ConfigApplication::setMappingFile
bool setMappingFile(const std::string &mapFilename)
Definition ConfigApplication.cpp:109
mqtt::mqttintegrator::lib::Mqtt::updateSubscriptions
static mqtt::lib::admin::ReloadResult updateSubscriptions(bool mustReconnect)
Definition Mqtt.cpp:88
reportState
static void reportState(const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
Definition mqttbridge.cpp:179
startClient
static SocketClientT< mqtt::mqttintegrator::SocketContextFactory, Args... > startClient(const std::string &instanceName, const std::function< void(typename SocketClientT< mqtt::mqttintegrator::SocketContextFactory >::Config *)> &configurator, Args &&... args)
Definition mqttintegrator.cpp:114
mqtt::lib::admin::makeMappingAdminRouter
express::Router makeMappingAdminRouter(ConfigApplication *configApplication, const AdminOptions &opt, ReloadCallback onDeploy)
Definition MappingAdminRouter.cpp:67

References reportState(), and mqtt::mqttintegrator::lib::Mqtt::updateSubscriptions().

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 95 of file mqttintegrator.cpp.

95 {
96 switch (state) {
97 case core::socket::State::OK:
98 VLOG(1) << instanceName << ": connected to '" << socketAddress.toString() << "'";
99 break;
100 case core::socket::State::DISABLED:
101 VLOG(1) << instanceName << ": disabled";
102 break;
103 case core::socket::State::ERROR:
104 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
105 break;
106 case core::socket::State::FATAL:
107 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
108 break;
109 }
110}

Referenced by main().

Here is the caller graph for this function:

◆ startClient() [1/2]

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

Definition at line 114 of file mqttintegrator.cpp.

116 {
117 using Client = SocketClientT<mqtt::mqttintegrator::SocketContextFactory, Args...>;
118 using SocketAddress = typename Client::SocketAddress;
119
120 Client socketClient = core::socket::stream::Client<Client>(instanceName, configurator, std::forward<Args>(args)...);
121
122 socketClient.getConfig()->setRetry();
123 socketClient.getConfig()->setRetryBase(1);
124 socketClient.getConfig()->setReconnect();
125
126 socketClient.connect([instanceName](const SocketAddress& socketAddress, const core::socket::State& state) {
127 reportState(instanceName, socketAddress, state);
128 });
129
130 return socketClient;
131}
reportState
static void reportState(const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
Definition mqttintegrator.cpp:95

◆ startClient() [2/2]

template<typename HttpClient>
HttpClient startClient ( const std::string & name,
const std::function< void(typename HttpClient::Config *)> & configurator = nullptr )

Definition at line 134 of file mqttintegrator.cpp.

134 {
135 using SocketAddress = typename HttpClient::SocketAddress;
136
137 const HttpClient httpClient(
138 name,
139 [](const std::shared_ptr<web::http::client::MasterRequest>& req) {
140 const std::string connectionName = req->getSocketContext()->getSocketConnection()->getConnectionName();
141
142 req->set("Sec-WebSocket-Protocol", "mqtt");
143
144 req->upgrade(
145 "/ws",
146 "websocket",
147 [connectionName](bool success) {
148 VLOG(1) << connectionName << ": HTTP Upgrade (http -> websocket||"
149 << "mqtt" << ") start " << (success ? "success" : "failed");
150 },
151 []([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req,
152 [[maybe_unused]] const std::shared_ptr<web::http::client::Response>& res,
153 [[maybe_unused]] bool success) {
154 },
155 [connectionName](const std::shared_ptr<web::http::client::Request>&, const std::string& message) {
156 VLOG(1) << connectionName << ": Request parse error: " << message;
157 });
158 },
159 []([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req) {
160 VLOG(1) << "Session ended";
161 });
162
163 if (configurator != nullptr) {
164 configurator(httpClient.getConfig());
165 }
166
167 httpClient.getConfig()->setRetry();
168 httpClient.getConfig()->setRetryBase(1);
169 httpClient.getConfig()->setReconnect();
170
171 httpClient.connect([name](const SocketAddress& socketAddress, const core::socket::State& state) {
172 reportState(name, socketAddress, state);
173 });
174
175 return httpClient;
176}