open_source / mqtt-cpp

Commit 76d013731490e4332f1552001a1db810be0a97e2

Authored by Peter M. Groen
1 parent e9159973

Fix on connection

Inline Side-by-side
Showing 2 changed files with 69 additions and 19 deletions
debug_log.txt 0 → 100644
  View file @76d0137
  1 +With broker present
  2 +
  3 +Jul 02 00:50:19 intelnuc64.osdev.nl test_connection[30797]: MQTT Client started[30797]: [MqttClient::MqttClient]
  4 +Jul 02 00:50:19 intelnuc64.osdev.nl [MqttClient::eventHandler][30797]: ConnectionTest - starting event handler.
  5 +Jul 02 00:50:19 intelnuc64.osdev.nl MqttClient[30797]: ConnectionTest - Request connect
  6 +Jul 02 00:50:19 intelnuc64.osdev.nl [ClientPaho][30797]: ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a - Setting the extra onConnected callback.
  7 +Jul 02 00:50:19 intelnuc64.osdev.nl [MqttClient::connectionStatusChanged][30797]: ConnectionTest - connection status of wrapped client ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a changed to 2
  8 +Jul 02 00:50:19 intelnuc64.osdev.nl ClientPaho[30797]: ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a - starting callback event handler
  9 +Jul 02 00:50:19 intelnuc64.osdev.nl [ClientPaho::onConnectSuccess][30797]: onConnectSucces triggered..
  10 +Jul 02 00:50:19 intelnuc64.osdev.nl [ClientPaho][30797]: onConnectSuccessOnInstance ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a - connected to endpoint localhost:1883 (mqtt version 4, session present FALSE )
  11 +Jul 02 00:50:19 intelnuc64.osdev.nl [MqttClient::connectionStatusChanged][30797]: ConnectionTest - connection status of wrapped client ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a changed to 4
  12 +Jul 02 00:50:30 intelnuc64.osdev.nl mosquitto[29463]: 1656715830: Client ConnectionTest_1_83746b87-431f-4ce1-82e3-2543e9b0a37a closed its connection.
  13 +
  14 +
  15 +Without a broker present
  16 +
  17 +Jul 02 00:55:33 intelnuc64.osdev.nl test_connection[31574]: MQTT Client started[31574]: [MqttClient::MqttClient]
  18 +Jul 02 00:55:33 intelnuc64.osdev.nl [MqttClient::eventHandler][31574]: ConnectionTest - starting event handler.
  19 +Jul 02 00:55:33 intelnuc64.osdev.nl MqttClient[31574]: ConnectionTest - Request connect
  20 +Jul 02 00:55:33 intelnuc64.osdev.nl [ClientPaho][31574]: ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 - Setting the extra onConnected callback.
  21 +Jul 02 00:55:33 intelnuc64.osdev.nl [MqttClient::connectionStatusChanged][31574]: ConnectionTest - connection status of wrapped client ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 changed to 2
  22 +Jul 02 00:55:33 intelnuc64.osdev.nl ClientPaho[31574]: ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 - starting callback event handler
  23 +Jul 02 00:55:33 intelnuc64.osdev.nl ClientPaho[31574]: onConnectFailureOnInstanceConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 - connection failed with code MQTTASYNC_FAILURE (TCP/TLS connect failure)
  24 +Jul 02 00:55:33 intelnuc64.osdev.nl [MqttClient::connectionStatusChanged][31574]: ConnectionTest - connection status of wrapped client ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 changed to 0
  25 +
  26 +Jul 02 00:55:42 intelnuc64.osdev.nl systemd[1]: Starting Mosquitto MQTT Broker...
  27 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: mosquitto version 2.0.14 starting
  28 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: Config loaded from /etc/mosquitto/mosquitto.conf.
  29 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: Starting in local only mode. Connections will only be possible from clients running on this machine.
  30 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: Create a configuration file which defines a listener to allow remote access.
  31 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: For more details see https://mosquitto.org/documentation/authentication-methods/
  32 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: Opening ipv4 listen socket on port 1883.
  33 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: Opening ipv6 listen socket on port 1883.
  34 +Jul 02 00:55:42 intelnuc64.osdev.nl mosquitto[31610]: 1656716142: mosquitto version 2.0.14 running
  35 +Jul 02 00:55:45 intelnuc64.osdev.nl mosquitto[31610]: 1656716145: New connection from 127.0.0.1:42196 on port 1883.
  36 +Jul 02 00:55:45 intelnuc64.osdev.nl mosquitto[31610]: 1656716145: New client connected from 127.0.0.1:42196 as ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 (p2, c1, k5).
  37 +Jul 02 00:55:53 intelnuc64.osdev.nl mosquitto[31610]: 1656716153: Client ConnectionTest_1_9eaee6bb-325b-406a-ae70-b4f3e7a41268 closed its connection.
  38 +
  39 +
src/clientpaho.cpp
  View file @76d0137
@@ -123,37 +123,29 @@ ClientPaho::ClientPaho(const std::string& _endpoint, @@ -123,37 +123,29 @@ ClientPaho::ClientPaho(const std::string& _endpoint,
123 , m_callbackEventQueue(m_clientId) 123 , m_callbackEventQueue(m_clientId)
124 , m_workerThread() 124 , m_workerThread()
125 { 125 {
126 - if (0 == s_numberOfInstances++) { 126 + if (0 == s_numberOfInstances++)
  127 + {
127 MQTTAsync_setTraceCallback(&ClientPaho::onLogPaho); 128 MQTTAsync_setTraceCallback(&ClientPaho::onLogPaho);
128 } 129 }
129 - // MLOGIC_COMMON_DEBUG("ClientPaho", "%1 - ctor ClientPaho %2", m_clientId, this); 130 +
  131 + LogDebug( "[ClientPaho::ClientPaho]", std::string( " " + m_clientId + " - ctor ClientPaho ") );
130 parseEndpoint(_endpoint); 132 parseEndpoint(_endpoint);
  133 +
131 auto rc = MQTTAsync_create(&m_client, m_endpoint.c_str(), m_clientId.c_str(), MQTTCLIENT_PERSISTENCE_NONE, nullptr); 134 auto rc = MQTTAsync_create(&m_client, m_endpoint.c_str(), m_clientId.c_str(), MQTTCLIENT_PERSISTENCE_NONE, nullptr);
132 if (MQTTASYNC_SUCCESS == rc) 135 if (MQTTASYNC_SUCCESS == rc)
133 { 136 {
134 MQTTAsync_setCallbacks(m_client, reinterpret_cast<void*>(this), ClientPaho::onConnectionLost, ClientPaho::onMessageArrived, ClientPaho::onDeliveryComplete); 137 MQTTAsync_setCallbacks(m_client, reinterpret_cast<void*>(this), ClientPaho::onConnectionLost, ClientPaho::onMessageArrived, ClientPaho::onDeliveryComplete);
135 - LogDebug( "[ClientPaho]", std::string( m_clientId + " - Setting the extra onConnected callback.") );  
136 -  
137 - auto ccb = MQTTAsync_setConnected( m_client, reinterpret_cast<void*>(this), ClientPaho::onFirstConnect );  
138 - if( MQTTASYNC_SUCCESS == ccb )  
139 - {  
140 - LogDebug( "[ClientPaho]", std::string( m_clientId + " - Setting the extra onConnected callback SUCCEEDED.") );  
141 - }  
142 - else  
143 - {  
144 - LogDebug( "[ClientPaho]", std::string( m_clientId + " - Setting the extra onConnected callback FAILED.") );  
145 - }  
146 -  
147 m_workerThread = std::thread(&ClientPaho::callbackEventHandler, this); 138 m_workerThread = std::thread(&ClientPaho::callbackEventHandler, this);
148 } 139 }
149 else 140 else
150 { 141 {
151 - // Do something sensible here. 142 + LogError( "[ClientPaho::ClientPaho]", std::string( m_clientId + " - Failed to create client for endpoint " + m_endpoint + ", return code " + pahoAsyncErrorCodeToString( rc ) ) );
152 } 143 }
153 } 144 }
154 145
155 ClientPaho::~ClientPaho() 146 ClientPaho::~ClientPaho()
156 { 147 {
  148 + LogDebug( "[ClientPaho::~ClientPaho]", std::string( m_clientId + " - dtor ClientPao" ) );
157 if( MQTTAsync_isConnected( m_client ) ) 149 if( MQTTAsync_isConnected( m_client ) )
158 { 150 {
159 this->unsubscribeAll(); 151 this->unsubscribeAll();
@@ -206,6 +198,8 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt ) @@ -206,6 +198,8 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt )
206 setConnectionStatus( ConnectionStatus::ConnectInProgress ); 198 setConnectionStatus( ConnectionStatus::ConnectInProgress );
207 } 199 }
208 200
  201 + LogInfo( "[ClientPaho::connect]", std::string( m_clientId + " - start connect to endpoint " + m_endpoint ) );
  202 +
209 MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer; 203 MQTTAsync_connectOptions conn_opts = MQTTAsync_connectOptions_initializer;
210 conn_opts.keepAliveInterval = 5; 204 conn_opts.keepAliveInterval = 5;
211 conn_opts.cleansession = 1; 205 conn_opts.cleansession = 1;
@@ -214,6 +208,18 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt ) @@ -214,6 +208,18 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt )
214 conn_opts.context = this; 208 conn_opts.context = this;
215 conn_opts.automaticReconnect = 1; 209 conn_opts.automaticReconnect = 1;
216 210
  211 + // Make sure we get a signal if the promise is fulfilled
  212 + auto ccb = MQTTAsync_setConnected( m_client, reinterpret_cast<void*>(this), ClientPaho::onFirstConnect );
  213 + if( MQTTASYNC_SUCCESS == ccb )
  214 + {
  215 + LogDebug( "[ClientPaho]", std::string( m_clientId + " - Setting the extra onConnected callback SUCCEEDED.") );
  216 + }
  217 + else
  218 + {
  219 + LogDebug( "[ClientPaho]", std::string( m_clientId + " - Setting the extra onConnected callback FAILED.") );
  220 + }
  221 +
  222 + // Setup the last will and testament, if so desired.
217 if( !lwt.topic().empty() ) 223 if( !lwt.topic().empty() )
218 { 224 {
219 MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer; 225 MQTTAsync_willOptions will_opts = MQTTAsync_willOptions_initializer;
@@ -221,6 +227,8 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt ) @@ -221,6 +227,8 @@ std::int32_t ClientPaho::connect( bool wait, const mqtt_LWT &amp;lwt )
221 will_opts.topicName = lwt.topic().c_str(); 227 will_opts.topicName = lwt.topic().c_str();
222 228
223 conn_opts.will = &will_opts; 229 conn_opts.will = &will_opts;
  230 +
  231 + LogDebug( "[ClientPaho::connect]", std::string( m_clientId + " - Set Last will and testament. Topic : " + lwt.topic() + " => Message : " + lwt.message() ) );
224 } 232 }
225 else 233 else
226 { 234 {
@@ -309,8 +317,10 @@ std::int32_t ClientPaho::disconnect(bool wait, int timeoutMs) @@ -309,8 +317,10 @@ std::int32_t ClientPaho::disconnect(bool wait, int timeoutMs)
309 } 317 }
310 318
311 int rc = MQTTAsync_disconnect(m_client, &disconn_opts); 319 int rc = MQTTAsync_disconnect(m_client, &disconn_opts);
312 - if (MQTTASYNC_SUCCESS != rc) {  
313 - if (MQTTASYNC_DISCONNECTED == rc) { 320 + if (MQTTASYNC_SUCCESS != rc)
  321 + {
  322 + if (MQTTASYNC_DISCONNECTED == rc)
  323 + {
314 currentStatus = ConnectionStatus::Disconnected; 324 currentStatus = ConnectionStatus::Disconnected;
315 } 325 }
316 326
@@ -1140,9 +1150,10 @@ void ClientPaho::onConnectSuccess(void* context, MQTTAsync_successData* response @@ -1140,9 +1150,10 @@ void ClientPaho::onConnectSuccess(void* context, MQTTAsync_successData* response
1140 if (context) 1150 if (context)
1141 { 1151 {
1142 auto* cl = reinterpret_cast<ClientPaho*>(context); 1152 auto* cl = reinterpret_cast<ClientPaho*>(context);
1143 - if (!response) { 1153 + if (!response)
  1154 + {
1144 // connect should always have a valid response struct. 1155 // connect should always have a valid response struct.
1145 - // ("ClientPaho", "onConnectSuccess - no response data"); 1156 + LogError( "[ClientPaho]", "onConnectSuccess - no response data");
1146 } 1157 }
1147 MqttSuccess resp(response->token, ConnectionData(response->alt.connect.serverURI, response->alt.connect.MQTTVersion, response->alt.connect.sessionPresent)); 1158 MqttSuccess resp(response->token, ConnectionData(response->alt.connect.serverURI, response->alt.connect.MQTTVersion, response->alt.connect.sessionPresent));
1148 cl->pushIncomingEvent([cl, resp]() { cl->onConnectSuccessOnInstance(resp); }); 1159 cl->pushIncomingEvent([cl, resp]() { cl->onConnectSuccessOnInstance(resp); });