MQTTSuite
Loading...
Searching...
No Matches
mqttstore.cpp File Reference
#include "SocketContextFactory.h"
#include "config.h"
#include "lib/ConfigSections.h"
#include <core/SNodeC.h>
#include <net/config/ConfigInstance.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/client/ConfigHTTP.h>
#include <web/http/http_utils.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 dependency graph for mqttstore.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)
static void logResponse (const std::shared_ptr< web::http::client::Request > &req, const std::shared_ptr< web::http::client::Response > &res)
template<template< typename SocketContextFactoryT, typename... ArgsT > typename SocketClient>
static SocketClient< mqtt::mqttstore::SocketContextFactory > startClient (const std::string &instanceName, const std::function< void(typename SocketClient< mqtt::mqttstore::SocketContextFactory >::Config *)> &configurator)
template<typename HttpClient>
static HttpClient startClient (const std::string &name, const std::function< void(typename HttpClient::Config *)> &configurator)
static void createConfig (net::config::ConfigInstance *config)
static void createWSConfig (net::config::ConfigInstance *config)
static int run (int argc, char *argv[])
int main (int argc, char *argv[])

Function Documentation

◆ createConfig()

void createConfig ( net::config::ConfigInstance * config)
static

Definition at line 163 of file mqttstore.cpp.

163 {
164 config //
165 ->newSubCommand<mqtt::mqttstore::lib::ConfigSession>();
166
167 config //
168 ->newSubCommand<mqtt::mqttstore::lib::ConfigSubscribe>();
169
170 config //
171 ->newSubCommand<mqtt::mqttstore::lib::ConfigDatabase>()
172 ->newSubCommand<mqtt::mqttstore::lib::ConfigStorage>();
173}

Referenced by createWSConfig().

Here is the caller graph for this function:

◆ createWSConfig()

void createWSConfig ( net::config::ConfigInstance * config)
static

Definition at line 175 of file mqttstore.cpp.

175 {
176 createConfig(config);
177
178 config->getSubCommand<web::http::client::ConfigHTTP>()
179 ->addOption("--target", "Websocket endpoint", "string", "/ws", CLI::TypeValidator<std::string>())
180 ->configurable();
181}
createConfig
static void createConfig(net::config::ConfigInstance *config)
Definition mqttstore.cpp:163

References createConfig().

Here is the call graph for this function:

◆ logResponse()

void logResponse ( const std::shared_ptr< web::http::client::Request > & req,
const std::shared_ptr< web::http::client::Response > & res )
static

Definition at line 75 of file mqttstore.cpp.

75 {
76 VLOG(1) << req->getConnectionName() << " HTTP response for: " << req->method << " " << req->url << " HTTP/" << req->httpMajor << "."
77 << req->httpMinor << "\n"
78 << httputils::toString(req->method,
79 req->url,
80 "HTTP/" + std::to_string(req->httpMajor) + "." + std::to_string(req->httpMinor),
81 req->getQueries(),
82 req->getHeaders(),
83 req->getTrailer(),
84 req->getCookies(),
85 {})
86 << "\n"
87 << httputils::toString(res->httpVersion, res->statusCode, res->reason, res->headers, res->cookies, res->body);
88}

◆ main()

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

Definition at line 277 of file mqttstore.cpp.

277 {
278 try {
279 return run(argc, argv);
280 } catch (const std::exception& exception) {
281 std::cerr << "mqttstore: fatal startup error: " << exception.what() << std::endl;
282 return EXIT_FAILURE;
283 } catch (...) {
284 std::cerr << "mqttstore: fatal startup error: unknown exception" << std::endl;
285 return EXIT_FAILURE;
286 }
287}
run
static int run(int argc, char *argv[])
Definition mqttstore.cpp:183

References run().

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 58 of file mqttstore.cpp.

58 {
59 switch (state) {
60 case core::socket::State::OK:
61 VLOG(1) << instanceName << ": connected to '" << socketAddress.toString() << "'";
62 break;
63 case core::socket::State::DISABLED:
64 VLOG(1) << instanceName << ": disabled";
65 break;
66 case core::socket::State::ERROR:
67 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
68 break;
69 case core::socket::State::FATAL:
70 VLOG(1) << instanceName << ": " << socketAddress.toString() << ": " << state.what();
71 break;
72 }
73}

◆ run()

int run ( int argc,
char * argv[] )
static

Definition at line 183 of file mqttstore.cpp.

183 {
184 core::SNodeC::init(argc, argv);
185
186#ifdef CONFIG_MQTTSUITE_STORE_TCP_IPV4
187 startClient<net::in::stream::legacy::SocketClient>("in-mqtt", [](net::in::stream::legacy::config::ConfigSocketClient* config) {
188 config->Remote::setPort(1883);
189 config->setDisableNagleAlgorithm();
190 createConfig(config);
191 });
192#endif
193
194#ifdef CONFIG_MQTTSUITE_STORE_TLS_IPV4
195 startClient<net::in::stream::tls::SocketClient>("in-mqtts", [](net::in::stream::tls::config::ConfigSocketClient* config) {
196 config->Remote::setPort(1883);
197 config->setDisableNagleAlgorithm();
198 createConfig(config);
199 });
200#endif
201
202#ifdef CONFIG_MQTTSUITE_STORE_TCP_IPV6
203 startClient<net::in6::stream::legacy::SocketClient>("in6-mqtt", [](net::in6::stream::legacy::config::ConfigSocketClient* config) {
204 config->Remote::setPort(1883);
205 config->setDisableNagleAlgorithm();
206 createConfig(config);
207 });
208#endif
209
210#ifdef CONFIG_MQTTSUITE_STORE_TLS_IPV6
211 startClient<net::in6::stream::tls::SocketClient>("in6-mqtts", [](net::in6::stream::tls::config::ConfigSocketClient* config) {
212 config->Remote::setPort(1883);
213 config->setDisableNagleAlgorithm();
214 createConfig(config);
215 });
216#endif
217
218#ifdef CONFIG_MQTTSUITE_STORE_UNIX
219 startClient<net::un::stream::legacy::SocketClient>("un-mqtt", [](net::un::stream::legacy::config::ConfigSocketClient* config) {
220 createConfig(config);
221 });
222#endif
223
224#ifdef CONFIG_MQTTSUITE_STORE_UNIX_TLS
225 startClient<net::un::stream::tls::SocketClient>("un-mqtts", [](net::un::stream::tls::config::ConfigSocketClient* config) {
226 createConfig(config);
227 });
228#endif
229
230#if defined(CONFIG_MQTTSUITE_STORE_TCP_IPV4) && defined(CONFIG_MQTTSUITE_STORE_WS)
231 startClient<web::http::legacy::in::Client>("in-wsmqtt", [](net::in::stream::legacy::config::ConfigSocketClient* config) {
232 config->Remote::setPort(8080);
233 config->setDisableNagleAlgorithm();
234 createWSConfig(config);
235 });
236#endif
237
238#if defined(CONFIG_MQTTSUITE_STORE_TLS_IPV4) && defined(CONFIG_MQTTSUITE_STORE_WSS)
239 startClient<web::http::tls::in::Client>("in-wsmqtts", [](net::in::stream::tls::config::ConfigSocketClient* config) {
240 config->Remote::setPort(8088);
241 config->setDisableNagleAlgorithm();
242 createWSConfig(config);
243 });
244#endif
245
246#if defined(CONFIG_MQTTSUITE_STORE_TCP_IPV6) && defined(CONFIG_MQTTSUITE_STORE_WS)
247 startClient<web::http::legacy::in6::Client>("in6-wsmqtt", [](net::in6::stream::legacy::config::ConfigSocketClient* config) {
248 config->Remote::setPort(8080);
249 config->setDisableNagleAlgorithm();
250 createWSConfig(config);
251 });
252#endif
253
254#if defined(CONFIG_MQTTSUITE_STORE_TLS_IPV6) && defined(CONFIG_MQTTSUITE_STORE_WSS)
255 startClient<web::http::tls::in6::Client>("in6-wsmqtts", [](net::in6::stream::tls::config::ConfigSocketClient* config) {
256 config->Remote::setPort(8088);
257 config->setDisableNagleAlgorithm();
258 createWSConfig(config);
259 });
260#endif
261
262#if defined(CONFIG_MQTTSUITE_STORE_UNIX) && defined(CONFIG_MQTTSUITE_STORE_WS)
263 startClient<web::http::legacy::un::Client>("un-wsmqtt", [](net::un::stream::legacy::config::ConfigSocketClient* config) {
264 createWSConfig(config);
265 });
266#endif
267
268#if defined(CONFIG_MQTTSUITE_STORE_UNIX_TLS) && defined(CONFIG_MQTTSUITE_STORE_WSS)
269 startClient<web::http::tls::un::Client>("un-wsmqtts", [](net::un::stream::tls::config::ConfigSocketClient* config) {
270 createWSConfig(config);
271 });
272#endif
273
274 return core::SNodeC::start();
275}
startClient
static SocketClient< mqtt::mqttstore::SocketContextFactory > startClient(const std::string &instanceName, const std::function< void(typename SocketClient< mqtt::mqttstore::SocketContextFactory >::Config *)> &configurator)
Definition mqttstore.cpp:92
createWSConfig
static void createWSConfig(net::config::ConfigInstance *config)
Definition mqttstore.cpp:175

Referenced by main().

Here is the caller graph for this function:

◆ startClient() [1/2]

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

Definition at line 92 of file mqttstore.cpp.

93 {
94 using Client = SocketClient<mqtt::mqttstore::SocketContextFactory>;
95 using SocketAddress = typename Client::SocketAddress;
96
97 Client socketClient = core::socket::stream::Client<Client>(instanceName, configurator);
98
99 socketClient.getConfig()->setRetry();
100 socketClient.getConfig()->setRetryBase(1);
101 socketClient.getConfig()->setReconnect();
102 socketClient.getConfig()->setDisabled();
103
104 socketClient.connect([instanceName](const SocketAddress& socketAddress, const core::socket::State& state) {
105 reportState(instanceName, socketAddress, state);
106 });
107
108 return socketClient;
109}
reportState
static void reportState(const std::string &instanceName, const core::socket::SocketAddress &socketAddress, const core::socket::State &state)
Definition mqttstore.cpp:58

◆ startClient() [2/2]

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

Definition at line 112 of file mqttstore.cpp.

112 {
113 using SocketAddress = typename HttpClient::SocketAddress;
114
115 const HttpClient httpClient(
116 name,
117 [](const std::shared_ptr<web::http::client::MasterRequest>& req) {
118 const std::string connectionName = req->getSocketContext()->getSocketConnection()->getConnectionName();
119 const std::string target = req->getSocketContext()
120 ->getSocketConnection()
121 ->getConfigInstance()
122 ->getSubCommand<web::http::client::ConfigHTTP>()
123 ->getOption("--target")
124 ->as<std::string>();
125
126 req->set("Sec-WebSocket-Protocol", "mqtt");
127
128 req->upgrade(
129 target,
130 "websocket",
131 [connectionName](bool success) {
132 VLOG(1) << connectionName << ": HTTP Upgrade (http -> websocket||mqtt) start " << (success ? "success" : "failed");
133 },
134 [connectionName](const std::shared_ptr<web::http::client::Request>& req,
135 const std::shared_ptr<web::http::client::Response>& res,
136 bool success) {
137 logResponse(req, res);
138 VLOG(1) << connectionName << ": HTTP Upgrade " << (success ? "success" : "failed");
139 },
140 [connectionName](const std::shared_ptr<web::http::client::Request>& req, const std::string& message) {
141 VLOG(1) << connectionName << ": Response parse error: " << message;
142 VLOG(1) << " Request was: " << req->method << " " << req->url << " HTTP/" << req->httpMajor << "." << req->httpMinor;
143 });
144 },
145 []([[maybe_unused]] const std::shared_ptr<web::http::client::Request>& req) {
146 VLOG(1) << "Session ended";
147 });
148
149 configurator(httpClient.getConfig());
150
151 httpClient.getConfig()->setRetry();
152 httpClient.getConfig()->setRetryBase(1);
153 httpClient.getConfig()->setReconnect();
154 httpClient.getConfig()->setDisabled();
155
156 httpClient.connect([name](const SocketAddress& socketAddress, const core::socket::State& state) {
157 reportState(name, socketAddress, state);
158 });
159
160 return httpClient;
161}
logResponse
static void logResponse(const std::shared_ptr< web::http::client::Request > &req, const std::shared_ptr< web::http::client::Response > &res)
Definition mqttstore.cpp:75