PubSubClient Class

Description
Defines a class of MQTT implementation for Ameba.

Syntax
class PubSubClient

Members

PubSubClient::PubSubClient

Description
Constructs a PubSubClient object and, if applicable, sets server address, port, call-back function, data stream and wifi client.

Syntax
PubSubClient::PubSubClient();
PubSubClient::PubSubClient(Client& client);
PubSubClient::PubSubClient(IPAddress, uint16_t, Client& client);
PubSubClient::PubSubClient(IPAddress, uint16_t, Client& client, Stream&);
PubSubClient::PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
PubSubClient::PubSubClient(IPAddress, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);
PubSubClient::PubSubClient(uint8_t *, uint16_t, Client& client);
PubSubClient::PubSubClient(uint8_t *, uint16_t, Client& client, Stream&);
PubSubClient::PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
PubSubClient::PubSubClient(uint8_t *, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);
PubSubClient::PubSubClient(const char*, uint16_t, Client& client);
PubSubClient::PubSubClient(const char*, uint16_t, Client& client, Stream&);
PubSubClient::PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client);
PubSubClient::PubSubClient(const char*, uint16_t, MQTT_CALLBACK_SIGNATURE, Client& client, Stream&);

Parameters
client: the network client to use, for example WiFiClient
IPAddress: MQTT server address
port: port for MQTT, usually 1883 for unencrypted connection
MQTT_CALLBACK_SIGNATURE: call-back function for MQTT
Stream: a stream to write received messages to

Returns
The function returns nothing.

Example Code
Example: MQTT_Basic
This example demonstrates the basic capabilities of the library.
It connects to an MQTT server then:
publishes “hello world” to the topic “outTopic”
subscribes to the topic “inTopic”, printing out any messages it receives, assumes the received payloads are strings not binary

#include "WiFi.h"     
#include "PubSubClient.h"     
  
// Update these with values suitable for your network.  
char ssid[] = "yourNetwork";     // your network SSID (name)  
char pass[] = "secretPassword";  // your network password  
int status  = WL_IDLE_STATUS;    // the Wifi radio's status  
  
char mqttServer[]     = "test.mosquitto.org";  
char clientId[]       = "amebaClient";  
char publishTopic[]   = "outTopic";  
char publishPayload[] = "hello world";  
char subscribeTopic[] = "inTopic";  
  
void callback(char* topic, byte* payload, unsigned int length) {  
  Serial.print("Message arrived [");  
  Serial.print(topic);  
  Serial.print("] ");  
  for (int i=0;i<length;i++) {  
    Serial.print((char)payload[i]);  
  }  
  Serial.println();  
}  
  
WiFiClient wifiClient;  
PubSubClient client(wifiClient);  
  
void reconnect() {  
  // Loop until we're reconnected  
  while (!client.connected()) {  
    Serial.print("Attempting MQTT connection...");  
    // Attempt to connect  
    if (client.connect(clientId)) {  
      Serial.println("connected");  
      // Once connected, publish an announcement...  
      client.publish(publishTopic, publishPayload);  
      // ... and resubscribe  
      client.subscribe(subscribeTopic);  
    } else {  
      Serial.print("failed, rc=");  
      Serial.print(client.state());  
      Serial.println(" try again in 5 seconds");  
      // Wait 5 seconds before retrying  
      delay(5000);  
    }  
  }  
}  
  
void setup()  
{  
  Serial.begin(38400);  
  
  while (status != WL_CONNECTED) {  
    Serial.print("Attempting to connect to SSID: ");  
    Serial.println(ssid);  
    // Connect to WPA/WPA2 network. Change this line if using open or WEP network:  
    status = WiFi.begin(ssid, pass);  
  
    // wait 10 seconds for connection:  
    delay(10000);  
  }  
  
  client.setServer(mqttServer, 1883);  
  client.setCallback(callback);  
  
  // Allow the hardware to sort itself out  
  delay(1500);  
}  
  
void loop()  
{  
  if (!client.connected()) {  
    reconnect();  
  }  
  client.loop();  
}  

Notes and Warnings
PubSubClient::PubSubClient(Client& client) would suffice for normal MQTT connection

PubSubClient::setServer

Description
Sets the server details.

Syntax
PubSubClient& PubSubClient::setServer(uint8_t * ip, uint16_t port)
PubSubClient& PubSubClient::setServer(IPAddress ip, uint16_t port)
PubSubClient& PubSubClient::setServer(const char * domain, uint16_t port)

Parameters
ip: the address of the server
port: the port to connect to, default 1883
domain: the address of the server

Returns
The client instance, allowing the function to be chained

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
NA

PubSubClient::setCallback

Description
Sets the message callback function.

Syntax
PubSubClient& PubSubClient::setCallback(MQTT_CALLBACK_SIGNATURE)

Parameters
MQTT_CALLBACK_SIGNATURE: a pointer to a message callback function called when a message arrives for a subscription created by this client.

Returns
The client instance, allowing the function to be chained.

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
NA

PubSubClient::setClient

Description
Sets the network client instance to use.

Syntax
PubSubClient& PubSubClient::setClient(Client& client)

Parameters
client: the network client to use, for example WiFiClient

Returns
The client instance, allowing the function to be chained

Example Code
NA

Notes and Warnings
NA

PubSubClient::setStream

Description
Sets the stream to write received messages to.

Syntax
PubSubClient& PubSubClient::setStream(Stream& stream)

Parameters
stream: a stream to write received messages to

Returns
The client instance, allowing the function to be chained.

Example Code
NA

Notes and Warnings
NA

PubSubClient::connect

Description
Connects the client to the server.

Syntax
boolean PubSubClient::connect(const char *id)
boolean PubSubClient::connect(const char *id, const char *user, const char *pass)
boolean PubSubClient::connect(const char *id, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage)
boolean PubSubClient::connect(const char *id, const char *user, const char *pass, const char* willTopic, uint8_t willQos, boolean willRetain, const char* willMessage)

Parameters
id: Client ID, a unique string identifier
user: Username for authentication, default NULL
pass: Password for authentication, default NULL
willTopic: the topic to be used by the will message
willQoS: the quality of service to be used by the will message
willRetain: whether the will should be published with the retain flag
willMessage: the payload of the will message

Returns
True – connection succeeded
False – connection failed

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
Client ID is required and should always be unique else connection might be rejected by the server.

PubSubClient::disconnect

Description
Disconnect the client

Syntax
void PubSubClient::disconnect(void)

Parameters
The function requires no input parameter.

Returns
The function returns nothing.

Example Code
NA

Notes and Warnings
NA

PubSubClient::publish

Description
Publishes a message to the specified topic.

Syntax
boolean PubSubClient::publish(const char* topic, const char* payload)
boolean PubSubClient::publish(const char* topic, const char* payload, boolean retained)
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength)
boolean PubSubClient::publish(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained)

Parameters
topic: the topic to publish to
payload: the message to publish
plength: the length of the payload. Required if payload is a byte[]
retained: whether the message should be retained
– False – not retained
– True – retained

Returns
False – publish failed, either connection lost or message too large
True – publish succeeded

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
Default max packet size is 128 bytes.

PubSubClient::publish_P

Description
Publishes a message stored in PROGMEM to the specified topic.

Syntax
boolean PubSubClient::publish_P(const char* topic, const uint8_t* payload, unsigned int plength, boolean retained)

Parameters
topic: the topic to publish to
payload: the message to publish
plength: the length of the payload. Required if the payload is a byte[]
retained: whether the message should be retained
– false – not retained
– true – retained

Returns
False – publish failed, either connection lost or message too large
True – publish succeeded

Example Code
NA

Notes and Warnings
NA

PubSubClient::subscribe

Description
Subscribes to messages published to the specified topic.

Syntax
boolean PubSubClient::subscribe(const char* topic)
boolean PubSubClient::subscribe(const char* topic, uint8_t qos)

Parameters
topic: the topic to subscribe to
qos: the qos to subscribe at

Returns
False – sending the subscribe failed, either connection lost or message too large
True – sending the subscribe succeeded

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
NA

PubSubClient::unsubscribe

Description
Unsubscribes from the specified topic.

Syntax
boolean PubSubClient::unsubscribe(const char* topic)

Parameters
topic: the topic to unsubscribe to

Returns
False – sending the unsubscribe failed, either connection lost or message too large
True – sending the unsubscribe succeeded

Example Code
NA

Notes and Warnings
NA

PubSubClient::loop

Description
A must method called regularly to allow the client to process incoming messages and maintain its connection to the server.

Syntax
boolean PubSubClient::loop(void)

Parameters
The function requires no input parameter.

Returns
False – the client is no longer connected
True – the client is still connected

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
A required method that should not be blocked for too long.

PubSubClient::connected

Description
Checks whether the client is connected to the server.

Syntax
boolean PubSubClient::connected(void)

Parameters
The function requires no input parameter.

Returns
False – the client is not connected
True – the client is connected

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
NA

PubSubClient::state

Description
Returns the current state of the client. If a connection attempt fails, this can be used to get more information about the failure.
All of the values have corresponding constants defined in PubSubClient.h.

Syntax
int PubSubClient::state(void)

Parameters
The function requires no input parameter.

Returns
-4 : MQTT_CONNECTION_TIMEOUT – the server didn’t respond within the keepalive time
-3 : MQTT_CONNECTION_LOST – the network connection was broken
-2 : MQTT_CONNECT_FAILED – the network connection failed
-1 : MQTT_DISCONNECTED – the client is disconnected cleanly
0 : MQTT_CONNECTED – the client is connected
1 : MQTT_CONNECT_BAD_PROTOCOL – the server doesn’t support the requested version of MQTT
2 : MQTT_CONNECT_BAD_CLIENT_ID – the server rejected the client identifier
3 : MQTT_CONNECT_UNAVAILABLE – the server was unable to accept the connection
4 : MQTT_CONNECT_BAD_CREDENTIALS – the username/password were rejected
5 : MQTT_CONNECT_UNAUTHORIZED – the client was not authorized to connect

Example Code
Example: MQTT_Basic
The details of the code can be found in the previous section of PubSubClient::PubSubClient.

Notes and Warnings
NA

PubSubClient::waitForAck

Description
Wait for an ACK sent from a host.

Syntax
void PubSubClient::waitForAck(uint8_t enable)

Parameters
enable: whether or not to enable the function feature of “wait for Acknowledgement”.

Returns
The function returns nothing.

Example Code
Example: google_cloud

#include "WiFi.h"     
#include "PubSubClient.h"          
  
#define count   100  
  
extern "C"{     
  extern char *jwt_generator(const unsigned char *private_key, char *project_id, int expiration_seconds);  
}  
  
  
WiFiSSLClient wifiClient;  
PubSubClient client(wifiClient);  
  
  
char ssid[] = "ssid"; // your network SSID (name)  
char pass[] = "pass"; // your network password (use for WPA, or use as key for WEP)  
int keyIndex = 0;     // your network key Index number (needed only for WEP)  
  
char GOOGLE_MQTT_SERVER[] = "mqtt.googleapis.com";  
int  GOOGLE_MQTT_PORT = 8883;  
  
char project_id[] = "amebago-193913";  
char registry_id[] = "amebago-registry";  
char device_id[] = "amebago-rs256-device";  
  
char clientUser[] = "unused";  
char *clientPass;  
char *mqtt_id;  
char *pub_topic;  
char payload[64];  
/* root CA can be download here: 
 *  https://pki.google.com/GIAG2.crt 
 */  
char* rootCABuff  = \  
"-----BEGIN CERTIFICATE-----\n" \  
"MIIDujCCAqKgAwIBAgILBAAAAAABD4Ym5g0wDQYJKoZIhvcNAQEFBQAwTDEgMB4G\n" \  
"A1UECxMXR2xvYmFsU2lnbiBSb290IENBIC0gUjIxEzARBgNVBAoTCkdsb2JhbFNp\n" \  
"Z24xEzARBgNVBAMTCkdsb2JhbFNpZ24wHhcNMDYxMjE1MDgwMDAwWhcNMjExMjE1\n" \  
"MDgwMDAwWjBMMSAwHgYDVQQLExdHbG9iYWxTaWduIFJvb3QgQ0EgLSBSMjETMBEG\n" \  
"A1UEChMKR2xvYmFsU2lnbjETMBEGA1UEAxMKR2xvYmFsU2lnbjCCASIwDQYJKoZI\n" \  
"hvcNAQEBBQADggEPADCCAQoCggEBAKbPJA6+Lm8omUVCxKs+IVSbC9N/hHD6ErPL\n" \  
"v4dfxn+G07IwXNb9rfF73OX4YJYJkhD10FPe+3t+c4isUoh7SqbKSaZeqKeMWhG8\n" \  
"eoLrvozps6yWJQeXSpkqBy+0Hne/ig+1AnwblrjFuTosvNYSuetZfeLQBoZfXklq\n" \  
"tTleiDTsvHgMCJiEbKjNS7SgfQx5TfC4LcshytVsW33hoCmEofnTlEnLJGKRILzd\n" \  
"C9XZzPnqJworc5HGnRusyMvo4KD0L5CLTfuwNhv2GXqF4G3yYROIXJ/gkwpRl4pa\n" \  
"zq+r1feqCapgvdzZX99yqWATXgAByUr6P6TqBwMhAo6CygPCm48CAwEAAaOBnDCB\n" \  
"mTAOBgNVHQ8BAf8EBAMCAQYwDwYDVR0TAQH/BAUwAwEB/zAdBgNVHQ4EFgQUm+IH\n" \  
"V2ccHsBqBt5ZtJot39wZhi4wNgYDVR0fBC8wLTAroCmgJ4YlaHR0cDovL2NybC5n\n" \  
"bG9iYWxzaWduLm5ldC9yb290LXIyLmNybDAfBgNVHSMEGDAWgBSb4gdXZxwewGoG\n" \  
"3lm0mi3f3BmGLjANBgkqhkiG9w0BAQUFAAOCAQEAmYFThxxol4aR7OBKuEQLq4Gs\n" \  
"J0/WwbgcQ3izDJr86iw8bmEbTUsp9Z8FHSbBuOmDAGJFtqkIk7mpM0sYmsL4h4hO\n" \  
"291xNBrBVNpGP+DTKqttVCL1OmLNIG+6KYnX3ZHu01yiPqFbQfXf5WRDLenVOavS\n" \  
"ot+3i9DAgBkcRcAtjOj4LaR0VknFBbVPFd5uRHg5h6h+u/N5GJG79G+dwfCMNYxd\n" \  
"AfvDbbnvRG15RjF+Cv6pgsH/76tuIMRQyV+dTZsXjAzlAcmgQWpzU/qlULRuJQ/7\n" \  
"TBj0/VLZjmmx6BEP3ojY+x1J96relc8geMJgEtslQIxq/H5COEBkEveegeGTLg==\n" \  
"-----END CERTIFICATE-----\n";  
  
  
/* Fill your private.pem.key with LINE ENDING */  
char *privateKeyBuff = \  
"-----BEGIN PRIVATE KEY-----\n" \  
  
"MIIEvQIBADANBgkqhkiG9w0BAQEFAASCBKcwggSjAgEAAoIBAQC4Zmd6c4ahWRJ1\n" \  
"OkN/BhyusnJgfOdlr9P9417q3mdEkv9d8i0uBuoDUQCBj6ue1PfJ0OEtxwyZGDkH\n" \  
"UmP88tO88s9lSZ19D+Czv002XjFn6hUyzZa3hbSpSMD90iUOxLVeMsk4RQ7cOfRG\n" \  
"ToKL2t54KUc2GFlGQ8jlvP8gLFmlDjftAGphZNmNkwXPmvwov812CdaA3mZ+YBMj\n" \  
"PLFDRYjE5bt2uf2+Cm57VRmPb8ONuAaTPgsLRRNGj+GJOTweMKV54TEPga0u7101\n" \  
"LqrofShdNMnMS4ebGzlK0RPaPxczdjHrLC9jHrG0bJDjSJRq4YaVx/zXbzbRribt\n" \  
"63NUkCT1AgMBAAECggEAXa/hqSwi3b0UjKzSeCoRzoxpYi5zno1rxpWLtwbSLtwE\n" \  
"lKWjYLwwjwjLmgf1qRgI4OeYUJrOAsZ0ywyIMo7pFxnCV3LEajLz9j8eqp3GukYL\n" \  
"CSm9BncPJ+cH7q2jGFLG1xo0c7taZnenbUUcPJQx7ZkDTi+mw/VSj66rbJw724h1\n" \  
"S+oeDbwEKv3ax1JA19iGOeL736d69EegV0+4cFC50gcxyVkwyfRapx0BYOPbUOo4\n" \  
"z0vpSeUKzBqKHmRq2O7g5F5TU2kopOVvVp9HbbG4UGah1BHDzwaiP409B/xmvAOm\n" \  
"q2rm3GuN5KX81eFkBlDjomK0GjopeX01Xla4a0VL7QKBgQDrwc2dgUO1Jdv6RU3L\n" \  
"7onbPxev6tD8I1H4toexOVg7/2x4C4WS/g7lCED2ORyKZgqpIw0IIS8OguMIm9Mg\n" \  
"AaUJQpqSpshFq2slkMPEvL8vt+BKsYKz36xFywijcG4zV3GZELTdVu0v/25++QMb\n" \  
"mOUgA5Pq9WKT+5qq5B5JQ8IdzwKBgQDIO7ezrovrT6hsbYgGu4DebeSk4s2DPLRg\n" \  
"AzzDdD2Ypl/DLhr5nzqdixmL//uW6qKQn52TErGIqkMzGtkz+TofVt+Yynh3xjRy\n" \  
"16P4NUBfsy9YMDcSWwpwWW35IFudGZ86bJZJgtRf+f0otlG6W7dcAVIqhlkXdM4e\n" \  
"/ULQU7Il+wKBgCx3ZFnFzMh4+JGuyqqhNj01HDmg94PnAYoAm31QzJSca5AE1E/S\n" \  
"PWrzcJVAVmLANliKdOXIpIB/LWUtRtftl3w0pMTuUi3Z1B7EvDf6RbExZEuSSY21\n" \  
"rV+ImPuCtDZY0uNE5GgvAhOggO3P98cXwneUVSzm1Y4F0blTx2aYMh+/AoGAEa5M\n" \  
"Q16HVmj7S0/Esit+bqWvievJD+ydVNkUVYH/KmqOjDKXCTHJQD4XLGiXM7VWU4T0\n" \  
"qhb9fD7kni+hvFgmjLvkFJ7UUmc7HGT0QqeZHpo49QWU51cIrfEHp/b2gAHSMJuE\n" \  
"DcuyqyLs+tpWjykoIMSxF7YzScHzrYLZkoHBel8CgYEAja6UGqcjC78mB4KErbCg\n" \  
"zA7EvYUmAiRdBkAxMvNDj54YIdIijP0hadJzcDwy2S8A3ckWJglMLxSJ1ghc5v/K\n" \  
"SygvyhLc5h5PTSNnElKr//7OLXPYalF/ACcwCum/Ppi6UJcHKBSW+ZGnt+Akennb\n" \  
"+BYcrelb7Sulr2Q5EBU6f1k=\n" \  
"-----END PRIVATE KEY-----\n";  
  
void setup() {  
    
  // attempt to connect to Wifi network:  
  while (WiFi.begin(ssid, pass) != WL_CONNECTED) {  
    delay(1000);  
  }  
  Serial.println("Connected to wifi");  
    
  Serial.println("Init MQTT");  
  wifiClient.setRootCA((unsigned char*)rootCABuff);        
  wifiClient.setClientCertificate(NULL, (unsigned char*)privateKeyBuff);  
	  
}  
	  
void loop() {  
  boolean ret = 0;    
    
  if (WiFi.status() != WL_CONNECTED) {  
    while (WiFi.begin(ssid, pass) != WL_CONNECTED)   
    {  
      delay(1000);  
    }  
    Serial.println("Connected to wifi");  
  }  
  
  if(ret == 0){        
    Serial.println("Ready for Publishing");  
    
    mqtt_id = (char *)malloc(strlen("projects/") + strlen(project_id) + strlen("/locations/us-central1/registries/")  
      + strlen(registry_id) + strlen("/devices/") + strlen(device_id) + 1);  
    sprintf(mqtt_id, "projects/%s/locations/us-central1/registries/%s/devices/%s", project_id, registry_id, device_id);  
        
    clientPass = jwt_generator((unsigned char*)privateKeyBuff, project_id, 3600*1);  
    
    pub_topic = (char *)malloc(strlen("/devices/") + strlen(device_id) + strlen("/events") + 1);  
    sprintf(pub_topic, "/devices/%s/events", device_id);      
      
    client.setServer(GOOGLE_MQTT_SERVER, GOOGLE_MQTT_PORT);  
    client.setPublishQos(MQTTQOS1);  
    client.waitForAck(true);  
  }  
    
  if (client.connect(mqtt_id, clientUser, clientPass) )  
  {   
      for(int i = 0; i < count; i++){  
        Serial.println("client connecting...");  
        memset(payload, 0x0, 64);  
        sprintf(payload, "This is Ameba's %d message!!", i);                       
        printf("Publishing the payload \"%s\" with len: %d\r\n", payload, strlen(payload));        
        ret = client.publish(pub_topic, payload);      
        if(ret == 1)  
          printf("client publish successfully!!! ret = %d\r\n",ret);              
        else{  
          printf("client publish unsuccessfully!!! ret = %d\r\n",ret);  
          break;  
        }                       
        if (!client.connected()) {  
          Serial.println("MQTT disconnect with server");  
          break;  
        }  
  
        delay(1000);  
      }  
      client.disconnect();    
  }  
    
  free(mqtt_id);  
  free(pub_topic);  
    
}  

Notes and Warnings
The parameter should be a Boolean value either True or False.

PubSubClient::setPublishQos

Description
Set Quality of Service to publish.

Syntax
uint8_t PubSubClient::setPublishQos(uint8_t qos_level)

Parameters
qos_level: Quality of Service either 0, 1 or 2

Returns
The QoS of publishing

Example Code
Example: google_cloud
The details of the code can be found in the previous section of PubSubClient::waitForAck.

Notes and Warnings
QoS can only be,
0: at most once
1: at least once
2: exactly once