open_source / mqtt-cpp

Commit 199d70751cd1e2b7fd66a5f1f044b62b982fc83b

Authored by Peter M. Groen
1 parent e8b2b36e

implement logging to mqttclient

Inline Side-by-side
Showing 6 changed files with 41 additions and 77 deletions
CMakeLists.txt
  View file @199d707
1 1 cmake_minimum_required(VERSION 3.0)
2 2 project(osdev_mqtt)
3   -# ==============================================================================
4   -# Check to see if we're a submodule or top-repo.
5   -if(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/cmake)
6   - message( STATUS "Looks like we're a single module" )
7   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/cmake)
8   -elseif(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
9   - message( STATUS "Looks like we're a submodule" )
10   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
11   -else()
12   - message( FATAL_ERROR "No cmake directory found. Did you run the submodules script?" )
13   -endif()
14   -
15   -# ==============================================================================
16   -# Check to see if there is versioning information available
17   -if(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/versioning)
18   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/versioning/cmake)
19   -elseif(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../versioning)
20   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../versioning/cmake)
21   -else()
22   - message( FATAL_ERROR "No ${CURRENT_SOURCE_DIR}/osdev_versioning directory found. Did you run the submodules script?" )
23   -endif()
  3 +LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/submodules/cmake)
24 4  
25 5 # ==============================================================================
26 6 # = Include build information
27   -include(osdevversion)
28 7 include(projectheader)
29   -
30 8 project_header(osdev_mqtt)
31 9  
  10 +add_subdirectory(submodules/logger)
32 11 add_subdirectory(src)
33 12 add_subdirectory(examples/pub)
34 13 add_subdirectory(examples/sub)
35 14  
36 15 include(packaging)
37 16 package_component()
38   -
39   -# add_subdirectory(test)
... ...
examples/pub/CMakeLists.txt
  View file @199d707
1 1 cmake_minimum_required(VERSION 3.0)
2   -LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../cmake)
  2 +LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/submodules/cmake)
3 3  
4 4 include(projectheader)
5 5 project_header(test_mqtt_pub)
... ... @@ -27,7 +27,7 @@ target_link_libraries(
27 27 set_target_properties( ${PROJECT_NAME} PROPERTIES
28 28 RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin
29 29 LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib
30   - ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/archive
  30 + ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib
31 31 )
32 32  
33 33 include(installation)
... ...
examples/sub/CMakeLists.txt
  View file @199d707
1 1 cmake_minimum_required(VERSION 3.0)
2   -LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../cmake)
  2 +LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/submodules/cmake)
3 3  
4 4 include(projectheader)
5 5 project_header(test_mqtt_sub)
... ...
src/CMakeLists.txt
  View file @199d707
1 1 cmake_minimum_required(VERSION 3.12)
2   -# ==============================================================================
3   -# Check to see if we're a submodule or top-repo.
4   -if(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
5   - message( STATUS "Looks like we're a single module" )
6   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../cmake)
7   -elseif(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../../cmake)
8   - message( STATUS "Looks like we're a submodule" )
9   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../cmake)
10   -else()
11   - message( FATAL_ERROR "No cmake directory found. Did you run the submodules script?" )
12   -endif()
  2 +LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_SOURCE_DIR}/submodules/cmake)
13 3  
14   -# ==============================================================================
15   -# Check to see if there is versioning information available
16   -if(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../versioning)
17   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../versioning/cmake)
18   -elseif(IS_DIRECTORY ${CMAKE_CURRENT_SOURCE_DIR}/../../versioning)
19   - LIST(APPEND CMAKE_MODULE_PATH ${CMAKE_CURRENT_SOURCE_DIR}/../../versioning/cmake)
20   -else()
21   - message( FATAL_ERROR "No ${CURRENT_SOURCE_DIR}/osdev_versioning directory found. Did you run the submodules script?" )
22   -endif()
23   -# ==============================================================================
24 4 include(projectheader)
25   -
26 5 project_header(mqtt-cpp)
27 6  
28 7 find_package( Boost REQUIRED COMPONENTS regex )
... ... @@ -31,6 +10,7 @@ include(compiler)
31 10  
32 11 include_directories(
33 12 ${CMAKE_SOURCE_DIR}/include
  13 + ${CMAKE_SOURCE_DIR}/submodules/logger/src
34 14 )
35 15  
36 16 set(SRC_LIST
... ... @@ -67,6 +47,7 @@ add_libraries(
67 47 Boost::boost
68 48 Boost::regex
69 49 paho-mqtt3a
  50 + ${CMAKE_SOURCE_DIR}/build/lib/liblogger.a
70 51 )
71 52  
72 53 include(installation)
... ...
src/mqttclient.cpp
  View file @199d707
... ... @@ -21,6 +21,9 @@
21 21 * ***************************************************************************/
22 22 #include "mqttclient.h"
23 23  
  24 +// osdev::components::logger
  25 +#include "log.h"
  26 +
24 27 // osdev::components::mqtt
25 28 #include "clientpaho.h"
26 29 #include "mqttutil.h"
... ... @@ -32,8 +35,10 @@
32 35 // std
33 36 #include <numeric>
34 37 #include <iostream>
  38 +#include <string>
35 39  
36 40 using namespace osdev::components::mqtt;
  41 +using namespace osdev::components::log;
37 42  
38 43 namespace {
39 44 /**
... ... @@ -63,6 +68,8 @@ MqttClient::MqttClient(const std::string&amp; _clientId, const std::function&lt;void(co
63 68 , m_eventQueue(_clientId)
64 69 , m_workerThread( std::thread( &MqttClient::eventHandler, this ) )
65 70 {
  71 + Log::init( "mqtt-library" );
  72 + LogInfo( "MQTT Client started", "[MqttClient::MqttClient]");
66 73 }
67 74  
68 75 MqttClient::~MqttClient()
... ... @@ -73,16 +80,16 @@ MqttClient::~MqttClient()
73 80 decltype(m_principalClient) principalClient{};
74 81  
75 82 OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
76   - // LogDebug( "MqttClient", std::string( m_clientId + " - cleanup principal client" ) );
  83 + LogDebug( "MqttClient", std::string( m_clientId + " - cleanup principal client" ) );
77 84 m_principalClient.swap(principalClient);
78 85 }
79 86  
80   - // LogDebug( "MqttClient", std::string( m_clientId + " - dtor stop queue" ) );
  87 + LogDebug( "MqttClient", std::string( m_clientId + " - dtor stop queue" ) );
81 88 m_eventQueue.stop();
82 89 if (m_workerThread.joinable()) {
83 90 m_workerThread.join();
84 91 }
85   - // LogDebug( "MqttClient", std::string( m_clientId + " - dtor ready" ) );
  92 + LogDebug( "MqttClient", std::string( m_clientId + " - dtor ready" ) );
86 93 }
87 94  
88 95 std::string MqttClient::clientId() const
... ... @@ -133,7 +140,7 @@ void MqttClient::connect(const std::string&amp; host, int port, const Credentials&amp; c
133 140  
134 141 void MqttClient::connect(const std::string& _endpoint)
135 142 {
136   - // LogInfo( "MqttClient", std::string( m_clientId + " - Request connect" ) );
  143 + LogInfo( "MqttClient", std::string( m_clientId + " - Request connect" ) );
137 144  
138 145 OSDEV_COMPONENTS_LOCKGUARD(m_interfaceMutex);
139 146 IMqttClientImpl* client(nullptr);
... ... @@ -147,7 +154,7 @@ void MqttClient::connect(const std::string&amp; _endpoint)
147 154 }
148 155 else
149 156 {
150   - // LogError( "MqttClient", std::string( m_clientId + " - Cannot connect to different endpoint. Disconnect first." ) );
  157 + LogError( "MqttClient", std::string( m_clientId + " - Cannot connect to different endpoint. Disconnect first." ) );
151 158 return;
152 159 }
153 160 }
... ... @@ -167,7 +174,7 @@ void MqttClient::connect(const std::string&amp; _endpoint)
167 174  
168 175 void MqttClient::disconnect()
169 176 {
170   - // LogInfo( "MqttClient", std::string( m_clientId + " - Request disconnect" ) );
  177 + LogInfo( "MqttClient", std::string( m_clientId + " - Request disconnect" ) );
171 178 OSDEV_COMPONENTS_LOCKGUARD(m_interfaceMutex);
172 179  
173 180 decltype(m_additionalClients) additionalClients{};
... ... @@ -176,6 +183,7 @@ void MqttClient::disconnect()
176 183 OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
177 184 if (!m_principalClient || m_principalClient->connectionStatus() == ConnectionStatus::Disconnected || m_principalClient->connectionStatus() == ConnectionStatus::DisconnectInProgress)
178 185 {
  186 + LogDebug( "MqttClient", std::string( m_clientId + " - Principal client not connected" ) );
179 187 return;
180 188 }
181 189 m_additionalClients.swap( additionalClients );
... ... @@ -188,8 +196,7 @@ void MqttClient::disconnect()
188 196 }
189 197  
190 198  
191   - // LogDebug( "MqttClient", std::string( m_clientId + " - Unsubscribe and disconnect clients" ) );
192   - // DebugLogToFIle ("MqttClient", "%1 - Unsubscribe and disconnect clients", m_clientId);
  199 + LogDebug( "MqttClient", std::string( m_clientId + " - Unsubscribe and disconnect clients" ) );
193 200 for ( auto& cl : clients )
194 201 {
195 202 cl->unsubscribeAll();
... ... @@ -215,15 +222,15 @@ Token MqttClient::publish(const MqttMessage&amp; message, int qos)
215 222 {
216 223 if( !m_principalClient )
217 224 {
218   - std::cout << "Principal client not initialized" << std::endl;
  225 + LogInfo( "[MqttClient::publish]", std::string( "Principal client not initialized") );
219 226 }
220 227  
221 228 if( m_principalClient->connectionStatus() == ConnectionStatus::Disconnected )
222 229 {
223 230 std::cout << "Unable to publish, not connected.." << std::endl;
224 231 }
225   - // ErrorLogToFIle ("MqttClient", "%1 - Unable to publish, not connected", m_clientId);
226   - // Throw (MqttException, "Not connected");
  232 + LogError("MqttClient", std::string( m_clientId + " - Unable to publish, not connected" ) );
  233 +
227 234 return Token(m_clientId, -1);
228 235 }
229 236 client = m_principalClient.get();
... ... @@ -233,7 +240,7 @@ Token MqttClient::publish(const MqttMessage&amp; message, int qos)
233 240  
234 241 Token MqttClient::subscribe(const std::string& topic, int qos, const std::function<void(MqttMessage)>& cb)
235 242 {
236   - // DebugLogToFIle ("MqttClient", "%1 - Subscribe to topic %2 with qos %3", m_clientId, topic, qos);
  243 + LogDebug( "[MqttClient::subscribe]", std::string( m_clientId + " - Subscribe to topic " + topic ) );
237 244 // OSDEV_COMPONENTS_LOCKGUARD(m_interfaceMutex);
238 245 bool clientFound = false;
239 246 IMqttClientImpl* client(nullptr);
... ... @@ -241,7 +248,7 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
241 248 // OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
242 249 if (!m_principalClient || m_principalClient->connectionStatus() == ConnectionStatus::Disconnected)
243 250 {
244   - // ErrorLogToFIle ("MqttClient", "%1 - Unable to subscribe, not connected", m_clientId);
  251 + LogError("MqttClient", std::string( m_clientId + " - Unable to subscribe, not connected" ) );
245 252 // throw (?)(MqttException, "Not connected");
246 253 return Token(m_clientId, -1);
247 254 }
... ... @@ -259,7 +266,7 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
259 266 }
260 267 }
261 268 if (!clientFound) {
262   - // DebugLogToFIle ("MqttClient", "%1 - Creating new ClientPaho instance for subscription on topic %2", m_clientId, topic);
  269 + LogDebug("[MqttClient::subscribe]", std::string( m_clientId + " - Creating new ClientPaho instance for subscription on topic " + topic ) );
263 270 std::string derivedClientId(generateUniqueClientId(m_clientId, m_additionalClients.size() + 2)); // principal client is nr 1.
264 271 m_additionalClients.emplace_back(std::make_unique<ClientPaho>(
265 272 m_endpoint,
... ... @@ -277,13 +284,13 @@ Token MqttClient::subscribe(const std::string&amp; topic, int qos, const std::functi
277 284  
278 285 std::set<Token> MqttClient::unsubscribe(const std::string& topic, int qos)
279 286 {
280   - // DebugLogToFIle ("MqttClient", "%1 - Unsubscribe from topic %2 with qos %3", m_clientId, topic, qos);
  287 + LogDebug("[MqttClient::unsubscribe]", std::string( m_clientId + " - Unsubscribe from topic " + topic ) );
281 288 OSDEV_COMPONENTS_LOCKGUARD(m_interfaceMutex);
282 289 std::vector<IMqttClientImpl*> clients{};
283 290 {
284 291 OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
285 292 if (!m_principalClient || m_principalClient->connectionStatus() == ConnectionStatus::Disconnected) {
286   - // ErrorLogToFIle ("MqttClient", "%1 - Unable to unsubscribe, not connected", m_clientId);
  293 + LogError("[MqttClient::unsubscribe]", std::string( m_clientId + " - Unable to unsubscribe, not connected" ) );
287 294 // Throw (MqttException, "Not connected");
288 295 return std::set<Token>();
289 296 }
... ... @@ -366,9 +373,7 @@ std::string MqttClient::endpoint() const
366 373  
367 374 void MqttClient::connectionStatusChanged(const std::string& id, ConnectionStatus cs)
368 375 {
369   - (void)id;
370   - (void)cs;
371   - // DebugLogToFIle ("MqttClient", "%1 - connection status of wrapped client %2 changed to %3", m_clientId, id, cs);
  376 + LogDebug("[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - connection status of wrapped client " + id + " changed to " + std::to_string( static_cast<int>(cs) ) ) );
372 377 IMqttClientImpl* principalClient{ nullptr };
373 378 std::vector<IMqttClientImpl*> clients{};
374 379 std::vector<ConnectionStatus> connectionStates{};
... ... @@ -399,7 +404,7 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
399 404 cl->resubscribe();
400 405 }
401 406 catch (const std::exception& e) {
402   - // ErrorLogToFIle ("MqttClient", "%1 - resubscribe on wrapped client %2 in context of connection status change in wrapped client %3 failed : %4", m_clientId, cl->clientId(), id, e.what());
  407 + LogError("[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - resubscribe on wrapped client " + cl->clientId() + " in context of connection status change in wrapped client : " + id + " => FAILED : " + e.what() ) );
403 408 }
404 409 }
405 410 m_activeTokensCV.notify_all();
... ... @@ -414,7 +419,7 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
414 419 auto waitFor = std::chrono::milliseconds(1000);
415 420 if (!waitForCompletionInternalClients(clients, waitFor, std::set<Token>{})) {
416 421 if (std::accumulate(clients.begin(), clients.end(), false, [](bool hasPending, IMqttClientImpl* client) { return hasPending || client->hasPendingSubscriptions(); })) {
417   - // WarnLogToFIle ("MqttClient", "%1 - subscriptions are not recovered within timeout.", m_clientId)
  422 + LogWarning("[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - subscriptions are not recovered within timeout." ) );
418 423 }
419 424 }
420 425 if (principalClient) {
... ... @@ -422,7 +427,7 @@ void MqttClient::connectionStatusChanged(const std::string&amp; id, ConnectionStatus
422 427 principalClient->publishPending();
423 428 }
424 429 catch (const std::exception& e) {
425   - // ErrorLogToFIle ("MqttClient", "%1 - publishPending on wrapped client %2 failed : %3", m_clientId, principalClient->clientId(), e.what());
  430 + LogError( "[MqttClient::connectionStatusChanged]", std::string( m_clientId + " - publishPending on wrapped client " + principalClient->clientId() + " => FAILED " + e.what() ) );
426 431 }
427 432 }
428 433 }
... ... @@ -441,7 +446,7 @@ void MqttClient::deliveryComplete(const std::string&amp; _clientId, std::int32_t tok
441 446 OSDEV_COMPONENTS_LOCKGUARD(m_internalMutex);
442 447 if (!m_activeTokens.insert(t).second) {
443 448 // This should not happen. This means that some callback on the wrapper never came.
444   - // ErrorLogToFIle ("MqttClient", "%1 -deliveryComplete, token %1 is already active", m_clientId, t);
  449 + LogDebug("[MqttClient::deliveryComplete]", std::string( m_clientId + " - deliveryComplete, token is already active" ) );
445 450 }
446 451 }
447 452 this->pushEvent([this, t]() {
... ... @@ -538,8 +543,9 @@ void MqttClient::pushEvent(std::function&lt;void()&gt; ev)
538 543  
539 544 void MqttClient::eventHandler()
540 545 {
541   - // InfoLogToFIle ("MqttClient", "%1 - starting event handler", m_clientId);
542   - for (;;) {
  546 + LogInfo("[MqttClient::eventHandler]", std::string( m_clientId + " - starting event handler." ) );
  547 + for (;;)
  548 + {
543 549 std::vector<std::function<void()>> events;
544 550 if (!m_eventQueue.pop(events))
545 551 {
... ... @@ -550,5 +556,5 @@ void MqttClient::eventHandler()
550 556 ev();
551 557 }
552 558 }
553   - // InfoLogToFIle ("MqttClient", "%1 - leaving event handler", m_clientId);
  559 + LogInfo("[MqttClient::eventHandler]", std::string( m_clientId + " - leaving event handler." ) );
554 560 }
... ...
test/CMakeLists.txt
  View file @199d707
... ... @@ -27,7 +27,7 @@ target_link_libraries(
27 27 set_target_properties( ${PROJECT_NAME} PROPERTIES
28 28 RUNTIME_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/bin
29 29 LIBRARY_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib
30   - ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/archive
  30 + ARCHIVE_OUTPUT_DIRECTORY ${CMAKE_BINARY_DIR}/lib
31 31 )
32 32  
33 33 include(installation)
... ...