MQTT API

This Library is a high-performance [MQTT] library for IoT publish/subscribe communications.

The library supports MQTT version 3.1.1 and has the following features:

Supports connect, publish, subscribe, ping and disconnect messages.
Message quality of service for reliable delivery.
Retained messages.
TLS encryption with ALPN over port 443.
High message throughput with exceptionally low overhead.
Wait for delivery or acknowledgement options.
Auto reconnect on network failures.
Parallelism via fiber coroutines.

Extensions

Mqtt	MQTT Protocol.

Functions

Mqtt *	mqttAlloc(RSocket sock, cchar clientId, MqttEventProc proc)
	Allocate an MQTT object.
int	mqttConnect(Mqtt mq, cchar username, cchar *password, int flags, MqttWaitFlags waitFlags)
	Establish a session with the MQTT broker.
void	mqttFree(Mqtt *mq)
	Free an Mqtt instance.
cchar *	mqttGetError(struct Mqtt *mq)
	Returns an error message for error code, error.
bool	mqttIsConnected(Mqtt *mq)
	Return true if the MQTT instance is connected to a peer.
int	mqttMsgsInQueue(Mqtt *mq)
	Get the number of messages to send in the send queue.
int	mqttMsgsToSend(Mqtt *mq)
	Get the number of messages in the queue.
int	mqttPing(Mqtt *mq)
	Ping the broker.
int	mqttPublish(Mqtt mq, cvoid msg, ssize size, int qos, MqttWaitFlags waitFlags, cchar *topic, ...)
	Publish an application message to the MQTT broker.
int	mqttPublishRetained(Mqtt mq, cvoid msg, ssize size, int qos, MqttWaitFlags waitFlags, cchar *topic, ...)
	Publish a retained message to the MQTT broker.
void	mqttSetMessageSize(Mqtt *mq, int size)
	Set the maximum message size.
void	mqttSetWill(Mqtt mq, cchar topic, cvoid *msg, ssize length)
	Set the will and testament message.
int	mqttSubscribe(Mqtt mq, MqttCallback callback, int maxQos, MqttWaitFlags waitFlags, cchar topic, ...)
	Subscribe to a topic.
int	mqttUnsubscribe(Mqtt mq, cchar topic, MqttWaitFlags wait)
	Unsubscribe from a topic.

Typedefs

Mqtt	MQTT instance.
MqttCallback	Message receipt callback.
MqttEventProc	MQTt event callback.
MqttHdr	Fixed header of a packet.
MqttMsg	Mqtt message.
MqttRecv	A struct used to deserialize/interpret an incoming packet from the broker.

Defines

#define	MQTT_BUF_SIZE 4096
	Receive buffer size.
#define	MQTT_EVENT_CONNECT 1
	A new connection was established.
#define	MQTT_EVENT_DISCONNECT 2
	The connection was terminated.
#define	MQTT_EVENT_STOPPING 3
	The application is stopping.
#define	MQTT_INLINE_BUF_SIZE 128
	Size of inline buffer.
#define	MQTT_KEEP_ALIVE (20 * 60)
	Default connection keep alive time in seconds.
#define	MQTT_MAX_MESSAGE_SIZE 256 * 1024 * 1024
	Max message size.
#define	MQTT_MSG_TIMEOUT 30
	Default message timeout in seconds.
#define	MQTT_PROTOCOL_LEVEL 0x04
	Protocol version 3.1.1.
#define	MQTT_TOPIC_SIZE 128
	Max topic size.
#define	MQTT_PROTOCOL_LEVEL 0x04
	Protocol version 3.1.1.

Mqtt

MQTT Protocol.

API Stability:: Evolving.

Fields:

RBuf *	buf	I/O read buffer.
uint	connected	Mqtt is currently connected flag.
int	error	Mqtt error flag.
char *	errorMsg	Mqtt error message.
MqttMsg	head	Head of message queue.
char *	id	Client ID.
int	keepAlive	Keep alive duration in seconds.
time_t	lastSend	Time of last send in seconds.
int	mask	R library wait event mask.
int	maxMessage	Maximum message size.
int	msgTimeout	Message timeout for retransmit.
int	nextId	Next message ID.
MqttEventProc	proc	Notification event callback
RSocket *	sock	Underlying socket transport.
uint	subscribedApi	Reserved.
int	throttle	Throttle delay in msec.
Time	throttleLastPub	Time of last publish or throttle.
Time	throttleMark	Throttle sending until Time.
RList *	topics	List of subscribed topics.
char *	willMsg	Will and testament message.
ssize	willMsgSize	Size of will message.
char *	willTopic	Will and testament topic.

Mqtt * * mqttAlloc (RSocket *sock, cchar *clientId, MqttEventProc proc)

Allocate an MQTT object.

Description:: This layers an MQTT transport over an existing connected socket.

Parameters:

sock	A connected socket object created via rAllocSocket and rConnectSocket.
clientId	Unique client identifier string.
proc	Event notification callback procedure.

API Stability:: Evolving.

int mqttConnect (Mqtt *mq, cchar *username, cchar *password, int flags, MqttWaitFlags waitFlags)

Establish a session with the MQTT broker.

Parameters:

mq	The Mqtt object.
username	The username to use when establishing the session with the MQTT broker. Set to NULL if a username is not required.
password	The password to use when establishing the session with the MQTT broker. Set to NULL if a password is not required.
flags	Additional MqttConnectFlags to use when establishing the connection. These flags are for forcing the session to start clean: MQTT_CONNECT_CLEAN_SESSION, the QOS level to publish the will and testament messages with, and whether or not the broker should retain the will_message, MQTT_CONNECT_WILL_RETAIN.
waitFlags	Wait flags. Set to MQTT_WAIT_NONE, MQTT_WAIT_SENT or MQTT_WAIT_ACK.

Returns:: Zero if successful.

API Stability:: Evolving.

cchar * * mqttGetError (struct Mqtt *mq)

Returns an error message for error code, error.

Parameters:

mq	Mqtt object.

Returns:: The associated error message.

API Stability:: Evolving.

bool mqttIsConnected (Mqtt *mq)

Return true if the MQTT instance is connected to a peer.

Parameters:

mq	The MQTT mq.

Returns:: True if connected.

API Stability:: Prototype.

int mqttMsgsInQueue (Mqtt *mq)

Get the number of messages to send in the send queue.

Parameters:

mq	The MQTT mq.

Returns:: The number of messages in the queue to send.

API Stability:: Evolving.

int mqttMsgsToSend (Mqtt *mq)

Get the number of messages in the queue.

Parameters:

mq	The MQTT mq.

Returns:: The number of messages in the queue.

API Stability:: Evolving.

int mqttPing (Mqtt *mq)

Ping the broker.

Parameters:

mq	The MQTT mq.

Returns:: Zero if successful.

API Stability:: Evolving.

int mqttPublish (Mqtt *mq, cvoid *msg, ssize size, int qos, MqttWaitFlags waitFlags, cchar *topic, ...)

Publish an application message to the MQTT broker.

Parameters:

mq	The Mqtt object.
msg	The data to be published.
size	The size of application_message in bytes.
qos	Quality of service. 0, 1, or 2.
waitFlags	Wait flags.
topic	Printf style topic string.
...	Topic args.

Returns:: Zero if successful.

int mqttPublishRetained (Mqtt *mq, cvoid *msg, ssize size, int qos, MqttWaitFlags waitFlags, cchar *topic, ...)

Publish a retained message to the MQTT broker.

Parameters:

mq	The Mqtt object.
msg	The data to be published.
size	The size of application_message in bytes.
qos	Quality of service. 0, 1, or 2.
waitFlags	Wait flags.
topic	Printf style topic string.
...	Topic args.

Returns:: Zero if successful.

void mqttSetMessageSize (Mqtt *mq, int size)

Set the maximum message size.

Description:: AWS supports a smaller maximum message size.

Parameters:

mq	The MQTT mq.
size	The maximum message size.

API Stability:: Evolving.

void mqttSetWill (Mqtt *mq, cchar *topic, cvoid *msg, ssize length)

Set the will and testament message.

Parameters:

mq	The MQTT mq.
topic	Will message topic.
msg	Message to send.
length	Message size.

API Stability:: Evolving.

int mqttSubscribe (Mqtt *mq, MqttCallback callback, int maxQos, MqttWaitFlags waitFlags, cchar *topic, ...)

Subscribe to a topic.

Parameters:

mq	Mqtt object.
callback	Function to invoke on receipt of messages.
maxQos	Maximum quality of service message to receive.
waitFlags	Wait flags.
topic	Printf style topic string.
...	Topic args.

Returns:: Zero if successful.

API Stability:: Evolving.

int mqttUnsubscribe (Mqtt *mq, cchar *topic, MqttWaitFlags wait)

Unsubscribe from a topic.

Parameters:

mq	The MQTT mq.
topic	The name of the topic to unsubscribe from.
wait	Wait flags.

Returns:: Zero if successful.

API Stability:: Evolving.

Functions

void mqttFree (Mqtt *mq)

Free an Mqtt instance.

Parameters:

mq	Mqtt instance allocated via mqttAlloc

Typedefs

typedef void(* MqttCallback) (struct MqttRecv *resp).

Message receipt callback.

Parameters:

resp	Message received structure.

API Stability:: Evolving.

typedef void(* MqttEventProc) (struct Mqtt *mq, int event).

MQTt event callback.

Parameters:

mq	Mqtt object created via mqttAlloc
event	Event type, set to MQTT_EVENT_CONNECT, MQTT_EVENT_DISCONNECT or MQTT_EVENT_STOPPING.

API Stability:: Evolving.

Mqtt

MQTT instance.

API Stability:: Evolving.

Fields:

RBuf *	buf	I/O read buffer.
uint	connected	Mqtt is currently connected flag.
int	error	Mqtt error flag.
char *	errorMsg	Mqtt error message.
MqttMsg	head	Head of message queue.
char *	id	Client ID.
int	keepAlive	Keep alive duration in seconds.
time_t	lastSend	Time of last send in seconds.
int	mask	R library wait event mask.
int	maxMessage	Maximum message size.
int	msgTimeout	Message timeout for retransmit.
int	nextId	Next message ID.
MqttEventProc	proc	Notification event callback
RSocket *	sock	Underlying socket transport.
uint	subscribedApi	Reserved.
int	throttle	Throttle delay in msec.
Time	throttleLastPub	Time of last publish or throttle.
Time	throttleMark	Throttle sending until Time.
RList *	topics	List of subscribed topics.
char *	willMsg	Will and testament message.
ssize	willMsgSize	Size of will message.
char *	willTopic	Will and testament topic.

MqttHdr

Fixed header of a packet.

API Stability:: Evolving.

Fields:

int		flags	Packet control flags.
int		length	Size in of the variable portion after fixed header and packet length.

MqttMsg

Mqtt message.

API Stability:: Internal.

Fields:

uchar *	buf	External message text buffer for large messages.
uchar *	end	End of message.
uchar *	endbuf	End of message buffer.
RFiber *	fiber	Message fiber to process the message.
int	id	Message sequence ID.
uchar	inlineBuf[MQTT_INLINE_BUF_SIZE]	Inline message text buffer for small message efficiency.
struct MqttMsg *	next	Next message in the queue.
struct MqttMsg *	prev	Previous message in the queue.
int	qos	Message quality of service.
time_t	sent	Time in seconds the message was sent.
uchar *	start	Start of message.
MqttMsgState	state	Message send status.
MqttPacketType	type	Message packet type.
MqttWaitFlags	wait	Message wait for send flags.

MqttRecv

A struct used to deserialize/interpret an incoming packet from the broker.

API Stability:: Evolving.

Fields:

MqttConnCode	code	Connection response code.
cuchar *	codes	Array of return codes for subscribed topics.
char *	data	Published message.
int	dataSize	Size of data.
uchar	dup	Set to 0 on first attempt to send packet.
uint	hasSession	Connection using an existing session.
struct MqttHdr	hdr	MQTT message fixed header.
int	id	Message ID.
MqttTopic *	matched	Matched topic.
struct Mqtt *	mq	Message queue.
int	numCodes	Size of codes.
uchar	qos	Quality of service.
uchar	retain	Message is retained.
char *	topic	Topic string.