Add 'connected_callback' that is invoked when CONNACK is received

CONNACK signals that the connection with the server has been established

examples/reconnect_subscriber.c

@@ -35,6 +35,13 @@ struct reconnect_state_t {
  */
 void reconnect_client(struct mqtt_client* client, void **reconnect_state_vptr);
 
+
+/**
+ * @brief My connect callback. This is called when we receive a CONNACK from
+ * the broker. We use it to subscribe to our topics.
+ */
+void connect_client(struct mqtt_client* client, void **reconnect_state_vptr);
+
 /**
  * @brief The function will be called whenever a PUBLISH message is received.
  */
@@ -98,7 +105,7 @@ int main(int argc, const char *argv[])
  /* setup a client */
  struct mqtt_client client;
 
- mqtt_init_reconnect(&client,
+ mqtt_init_reconnect(&client, connect_client,
  reconnect_client, &reconnect_state,
  publish_callback
  );
@@ -165,6 +172,11 @@ void reconnect_client(struct mqtt_client* client, void **reconnect_state_vptr)
  uint8_t connect_flags = MQTT_CONNECT_CLEAN_SESSION;
  /* Send connection request to the broker. */
  mqtt_connect(client, client_id, NULL, NULL, 0, NULL, NULL, connect_flags, 400);
+}
+
+void connect_client(struct mqtt_client* client, void **reconnect_state_vptr)
+{
+ struct reconnect_state_t *reconnect_state = *((struct reconnect_state_t**) reconnect_state_vptr);
 
  /* Subscribe to the topic. */
  mqtt_subscribe(client, reconnect_state->topic, 0);

include/mqtt.h

@@ -1192,18 +1192,26 @@ struct mqtt_client {
  */
  enum MQTTErrors (*inspector_callback)(struct mqtt_client*);
 
+ /**
+ * @brief A callback that is called when the connection is established.
+ * 
+ * This callback is invoked when CONNACK is received, and can be used for
+ * things like session configurations (i.e. subscriptions). 
+ */
+ void (*connected_callback)(struct mqtt_client*, void**);
+
  /**
  * @brief A callback that is called whenever the client is in an error state.
  * 
  * This callback is responsible for: application level error handling, closing
- * previous sockets, and reestabilishing the connection to the broker and 
- * session configurations (i.e. subscriptions). 
+ * previous sockets, and reestabilishing the connection to the broker.
  */
  void (*reconnect_callback)(struct mqtt_client*, void**);
 
  /**
  * @brief A pointer to some state. A pointer to this member is passed to 
- * \ref mqtt_client.reconnect_callback.
+ * \ref mqtt_client.reconnect_callback and
+ * \ref mqtt_client.connected_callback.
  */
  void* reconnect_state;
 
@@ -1379,6 +1387,8 @@ enum MQTTErrors mqtt_init(struct mqtt_client *client,
  * @pre None.
  * 
  * @param[in,out] client The MQTT client that will be initialized.
+ * @param[in] connected_callback The callback that will be called when a connection has been
+ * established.
  * @param[in] reconnect_callback The callback that will be called to connect/reconnect the 
  * client to the broker and perform application level error handling. 
  * @param[in] reconnect_state A pointer to some state data for your \p reconnect_callback.
@@ -1396,6 +1406,7 @@ enum MQTTErrors mqtt_init(struct mqtt_client *client,
  *
  */
 void mqtt_init_reconnect(struct mqtt_client *client,
+ void (*connected_callback)(struct mqtt_client *client, void** state),
  void (*reconnect_callback)(struct mqtt_client *client, void** state),
  void *reconnect_state,
  void (*publish_response_callback)(void** state, struct mqtt_response_publish *publish));

src/mqtt.c

@@ -143,14 +143,16 @@ enum MQTTErrors mqtt_init(struct mqtt_client *client,
  client->send_offset = 0;
 
  client->inspector_callback = NULL;
+ client->connected_callback = NULL;
  client->reconnect_callback = NULL;
  client->reconnect_state = NULL;
 
  return MQTT_OK;
 }
 
 void mqtt_init_reconnect(struct mqtt_client *client,
- void (*reconnect)(struct mqtt_client *, void**),
+ void (*connected_callback)(struct mqtt_client *, void**),
+ void (*reconnect_callback)(struct mqtt_client *, void**),
  void *reconnect_state,
  void (*publish_response_callback)(void** state, struct mqtt_response_publish *publish))
 {
@@ -176,7 +178,8 @@ void mqtt_init_reconnect(struct mqtt_client *client,
  client->send_offset = 0;
 
  client->inspector_callback = NULL;
- client->reconnect_callback = reconnect;
+ client->connected_callback = connected_callback;
+ client->reconnect_callback = reconnect_callback;
  client->reconnect_state = reconnect_state;
 }
 
@@ -719,6 +722,7 @@ ssize_t __mqtt_recv(struct mqtt_client *client)
  MQTT_CONTROL_PINGRESP:
  -> release PINGREQ
  */
+ int has_connected = 0;
  switch (response.fixed_header.control_type) {
  case MQTT_CONTROL_CONNACK:
  /* release associated CONNECT */
@@ -741,6 +745,8 @@ ssize_t __mqtt_recv(struct mqtt_client *client)
  mqtt_recv_ret = MQTT_ERROR_CONNECTION_REFUSED;
  }
  break;
+ } else {
+ has_connected = 1;
  }
  break;
  case MQTT_CONTROL_PUBLISH:
@@ -890,10 +896,16 @@ ssize_t __mqtt_recv(struct mqtt_client *client)
  client->recv_buffer.curr -= consumed;
  client->recv_buffer.curr_sz += (unsigned long)consumed;
  }
+ if (has_connected && client->connected_callback != NULL) {
+ MQTT_PAL_MUTEX_UNLOCK(&client->mutex);
+ client->connected_callback(client, &client->reconnect_state);
+ MQTT_PAL_MUTEX_LOCK(&client->mutex);
+ }
  }
 
  /* In case there was some error handling the (well formed) message, we end up here */
  MQTT_PAL_MUTEX_UNLOCK(&client->mutex);
+
  return mqtt_recv_ret;
 }