MQTT client

Modules
	Options

Data Structures
struct	mqtt_connect_client_info_t

Macros
#define	MQTT_PORT   1883
#define	mqtt_subscribe(client, topic, qos, cb, arg)   mqtt_sub_unsub(client, topic, qos, cb, arg, 1)
#define	mqtt_unsubscribe(client, topic, cb, arg)   mqtt_sub_unsub(client, topic, 0, cb, arg, 0)

Typedefs
typedef void(*	mqtt_connection_cb_t) (mqtt_client_t *client, void *arg, mqtt_connection_status_t status)
typedef void(*	mqtt_incoming_data_cb_t) (void *arg, const u8_t *data, u16_t len, u8_t flags)
typedef void(*	mqtt_incoming_publish_cb_t) (void *arg, const char *topic, u32_t tot_len)
typedef void(*	mqtt_request_cb_t) (void *arg, err_t err)

Enumerations
enum	mqtt_connection_status_t
enum	{ MQTT_DATA_FLAG_LAST = 1 }

Functions
err_t	mqtt_publish (mqtt_client_t *client, const char *topic, const void *payload, u16_t payload_length, u8_t qos, u8_t retain, mqtt_request_cb_t cb, void *arg)
err_t	mqtt_sub_unsub (mqtt_client_t *client, const char *topic, u8_t qos, mqtt_request_cb_t cb, void *arg, u8_t sub)
void	mqtt_set_inpub_callback (mqtt_client_t *client, mqtt_incoming_publish_cb_t pub_cb, mqtt_incoming_data_cb_t data_cb, void *arg)
mqtt_client_t *	mqtt_client_new (void)
err_t	mqtt_client_connect (mqtt_client_t *client, const ip_addr_t *ip_addr, u16_t port, mqtt_connection_cb_t cb, void *arg, const struct mqtt_connect_client_info_t *client_info)
void	mqtt_disconnect (mqtt_client_t *client)
u8_t	mqtt_client_is_connected (mqtt_client_t *client)

Detailed Description

MQTT client for lwIP

Author: Erik Andersson

Details of the MQTT protocol can be found at:
http://docs.oasis-open.org/mqtt/mqtt/v3.1.1/os/mqtt-v3.1.1-os.html 

-----------------------------------------------------------------
1. Initial steps, reserve memory and make connection to server:

1.1: Provide storage

Static allocation:
  mqtt_client_t static_client;
  example_do_connect(&static_client);

Dynamic allocation:
  mqtt_client_t *client = mqtt_client_new();
  if(client != NULL) {
    example_do_connect(&client);
  }
  
1.2: Establish Connection with server

void example_do_connect(mqtt_client_t *client)
{
  struct mqtt_connect_client_info_t ci;
  err_t err;
  
  /* Setup an empty client info structure */
  memset(&ci, 0, sizeof(ci));
  
  /* Minimal amount of information required is client identifier, so set it here */ 
  ci.client_id = "lwip_test";
  
  /* Initiate client and connect to server, if this fails immediately an error code is returned
     otherwise mqtt_connection_cb will be called with connection result after attempting 
     to establish a connection with the server. 
     For now MQTT version 3.1.1 is always used */
  
  err = mqtt_client_connect(client, ip_addr, MQTT_PORT, mqtt_connection_cb, 0, &ci);
  
  /* For now just print the result code if something goes wrong
  if(err != ERR_OK) {
    printf("mqtt_connect return %d\n", err);
  }
}

Connection to server can also be probed by calling mqtt_client_is_connected(client) 

-----------------------------------------------------------------
2. Implementing the connection status callback


static void mqtt_connection_cb(mqtt_client_t *client, void *arg, mqtt_connection_status_t status)
{
  err_t err;
  if(status == MQTT_CONNECT_ACCEPTED) {
    printf("mqtt_connection_cb: Successfully connected\n");
    
    /* Setup callback for incoming publish requests */
    mqtt_set_inpub_callback(client, mqtt_incoming_publish_cb, mqtt_incoming_data_cb, arg);
    
    /* Subscribe to a topic named "subtopic" with QoS level 1, call mqtt_sub_request_cb with result */ 
    err = mqtt_subscribe(client, "subtopic", 1, mqtt_sub_request_cb, arg);

    if(err != ERR_OK) {
      printf("mqtt_subscribe return: %d\n", err);
    }
  } else {
    printf("mqtt_connection_cb: Disconnected, reason: %d\n", status);
    
    /* Its more nice to be connected, so try to reconnect */
    example_do_connect(client);
  }  
}

static void mqtt_sub_request_cb(void *arg, err_t result)
{
  /* Just print the result code here for simplicity, 
     normal behaviour would be to take some action if subscribe fails like 
     notifying user, retry subscribe or disconnect from server */
  printf("Subscribe result: %d\n", result);
}

-----------------------------------------------------------------
3. Implementing callbacks for incoming publish and data

/* The idea is to demultiplex topic and create some reference to be used in data callbacks
   Example here uses a global variable, better would be to use a member in arg
   If RAM and CPU budget allows it, the easiest implementation might be to just take a copy of
   the topic string and use it in mqtt_incoming_data_cb
*/
static int inpub_id;
static void mqtt_incoming_publish_cb(void *arg, const char *topic, u32_t tot_len)
{
  printf("Incoming publish at topic %s with total length %u\n", topic, (unsigned int)tot_len);

  /* Decode topic string into a user defined reference */
  if(strcmp(topic, "print_payload") == 0) {
    inpub_id = 0;
  } else if(topic[0] == 'A') {
    /* All topics starting with 'A' might be handled at the same way */
    inpub_id = 1;
  } else {
    /* For all other topics */
    inpub_id = 2;
  }
}

static void mqtt_incoming_data_cb(void *arg, const u8_t *data, u16_t len, u8_t flags)
{
  printf("Incoming publish payload with length %d, flags %u\n", len, (unsigned int)flags);

  if(flags & MQTT_DATA_FLAG_LAST) {
    /* Last fragment of payload received (or whole part if payload fits receive buffer
       See MQTT_VAR_HEADER_BUFFER_LEN)  */

    /* Call function or do action depending on reference, in this case inpub_id */
    if(inpub_id == 0) {
      /* Don't trust the publisher, check zero termination */
      if(data[len-1] == 0) {
        printf("mqtt_incoming_data_cb: %s\n", (const char *)data);
      }
    } else if(inpub_id == 1) {
      /* Call an 'A' function... */
    } else {
      printf("mqtt_incoming_data_cb: Ignoring payload...\n");
    }
  } else {
    /* Handle fragmented payload, store in buffer, write to file or whatever */
  }
}

-----------------------------------------------------------------
4. Using outgoing publish


void example_publish(mqtt_client_t *client, void *arg)
{
  const char *pub_payload= "PubSubHubLubJub";
  err_t err;
  u8_t qos = 2; /* 0 1 or 2, see MQTT specification */
  u8_t retain = 0; /* No don't retain such crappy payload... */
  err = mqtt_publish(client, "pub_topic", pub_payload, strlen(pub_payload), qos, retain, mqtt_pub_request_cb, arg);
  if(err != ERR_OK) {
    printf("Publish err: %d\n", err);
  }
}

/* Called when publish is complete either with sucess or failure */
static void mqtt_pub_request_cb(void *arg, err_t result)
{
  if(result != ERR_OK) {
    printf("Publish result: %d\n", result);
  }
}

-----------------------------------------------------------------
5. Disconnecting

Simply call mqtt_disconnect(client)

Macro Definition Documentation

MQTT_PORT

#define MQTT_PORT   1883

Default MQTT port

mqtt_subscribe

#define mqtt_subscribe	(		client,
			topic,
			qos,
			cb,
			arg
	)		mqtt_sub_unsub(client, topic, qos, cb, arg, 1)

Subscribe to topic

mqtt_unsubscribe

#define mqtt_unsubscribe	(		client,
			topic,
			cb,
			arg
	)		mqtt_sub_unsub(client, topic, 0, cb, arg, 0)

Unsubscribe to topic

Typedef Documentation

mqtt_connection_cb_t

typedef void(* mqtt_connection_cb_t) (mqtt_client_t *client, void *arg, mqtt_connection_status_t status)

Function prototype for mqtt connection status callback. Called when client has connected to the server after initiating a mqtt connection attempt by calling mqtt_connect() or when connection is closed by server or an error

Parameters

client	MQTT client itself
arg	Additional argument to pass to the callback function
status	Connect result code or disconnection notification

See also

mqtt_connection_status_t

mqtt_incoming_data_cb_t

typedef void(* mqtt_incoming_data_cb_t) (void *arg, const u8_t *data, u16_t len, u8_t flags)

Function prototype for MQTT incoming publish data callback function. Called when data arrives to a subscribed topic

See also

mqtt_subscribe

Parameters

arg	Additional argument to pass to the callback function
data	User data, pointed object, data may not be referenced after callback return, NULL is passed when all publish data are delivered
len	Length of publish data fragment
flags	MQTT_DATA_FLAG_LAST set when this call contains the last part of data from publish message

mqtt_incoming_publish_cb_t

typedef void(* mqtt_incoming_publish_cb_t) (void *arg, const char *topic, u32_t tot_len)

Function prototype for MQTT incoming publish function. Called when an incoming publish arrives to a subscribed topic

See also

mqtt_subscribe

Parameters

arg	Additional argument to pass to the callback function
topic	Zero terminated Topic text string, topic may not be referenced after callback return
tot_len	Total length of publish data, if set to 0 (no publish payload) data callback will not be invoked

mqtt_request_cb_t

typedef void(* mqtt_request_cb_t) (void *arg, err_t err)

Function prototype for mqtt request callback. Called when a subscribe, unsubscribe or publish request has completed

Parameters

arg	Pointer to user data supplied when invoking request
err	ERR_OK on success ERR_TIMEOUT if no response was received within timeout, ERR_ABRT if (un)subscribe was denied

Enumeration Type Documentation

anonymous enum

anonymous enum

Data callback flags

Enumerator
MQTT_DATA_FLAG_LAST	Flag set when last fragment of data arrives in data callback

mqtt_connection_status_t

enum mqtt_connection_status_t

Connection status codes

Function Documentation

mqtt_client_connect()

err_t mqtt_client_connect	(	mqtt_client_t *	client,
		const ip_addr_t *	ip_addr,
		u16_t	port,
		mqtt_connection_cb_t	cb,
		void *	arg,
		const struct mqtt_connect_client_info_t *	client_info
	)

Connect to MQTT server

Parameters

client	MQTT client
ip_addr	Server IP
port	Server port
cb	Connection state change callback
arg	User supplied argument to connection callback
client_info	Client identification and connection options

Returns

ERR_OK if successful,

See also

err_t enum for other results

mqtt_client_is_connected()

u8_t mqtt_client_is_connected

(

mqtt_client_t *

client

)

Check connection with server

Parameters

client

MQTT client

Returns

1 if connected to server, 0 otherwise

mqtt_client_new()

mqtt_client_t* mqtt_client_new

(

void

)

Create a new MQTT client instance

Returns

Pointer to instance on success, NULL otherwise

mqtt_disconnect()

void mqtt_disconnect

(

mqtt_client_t *

client

)

Disconnect from MQTT server

Parameters

client

MQTT client

mqtt_publish()

err_t mqtt_publish	(	mqtt_client_t *	client,
		const char *	topic,
		const void *	payload,
		u16_t	payload_length,
		u8_t	qos,
		u8_t	retain,
		mqtt_request_cb_t	cb,
		void *	arg
	)

MQTT publish function.

Parameters

client	MQTT client
topic	Publish topic string
payload	Data to publish (NULL is allowed)
payload_length	Length of payload (0 is allowed)
qos	Quality of service, 0 1 or 2
retain	MQTT retain flag
cb	Callback to call when publish is complete or has timed out
arg	User supplied argument to publish callback

Returns

ERR_OK if successful ERR_CONN if client is disconnected ERR_MEM if short on memory

mqtt_set_inpub_callback()

void mqtt_set_inpub_callback	(	mqtt_client_t *	client,
		mqtt_incoming_publish_cb_t	pub_cb,
		mqtt_incoming_data_cb_t	data_cb,
		void *	arg
	)

Set callback to handle incoming publish requests from server

Parameters

client	MQTT client
pub_cb	Callback invoked when publish starts, contain topic and total length of payload
data_cb	Callback for each fragment of payload that arrives
arg	User supplied argument to both callbacks

mqtt_sub_unsub()

err_t mqtt_sub_unsub	(	mqtt_client_t *	client,
		const char *	topic,
		u8_t	qos,
		mqtt_request_cb_t	cb,
		void *	arg,
		u8_t	sub
	)

MQTT subscribe/unsubscribe function.

Parameters

client	MQTT client
topic	topic to subscribe to
qos	Quality of service, 0 1 or 2 (only used for subscribe)
cb	Callback to call when subscribe/unsubscribe reponse is received
arg	User supplied argument to publish callback
sub	1 for subscribe, 0 for unsubscribe

Returns

ERR_OK if successful,

See also

err_t enum for other results