mqtt-redis

A library to use Redis as the persistence store for Paho Rust MQTT clients.

MQTT is a light-weight, distributed, messaging system particularly tailored for devices that have unreliable network connections. Part of the way it achieves a higher quality of service is that clients can use a persistence store to keep messages tha are in-flight until they are confirmed as received by the remote system. This way, even if the client application crashes and restarts, the messages can be confirmed or re-sent.

Typically messages are persisted to disk. On some smaller flash-based devices, such as Embedded Linux systems, this might not be the most efficient means to store the messages. A Redis server, running on the client device can serve as a convenient store for messages.

The Paho MQTT Rust Library

The Paho MQTT Rust library is a new wrapper around the Paho C library. As of this time it is still unpublished and in an incubator state. The source repository lives on GitHub at:

https://github.com/eclipse/paho.mqtt.rust

The Cargo.toml file lists a local address for the project. This should be updated to wherever the Git repository is cloned on your local system.

Rust Redis Client

This library uses the "redis" crate v0.8 to communicate with the Redis server. It is listed as a dependency in the Cargo.toml file. The project's home page can be found at:

https://github.com/mitsuhiko/redis-rs

Note that this client assumes that Redis is running on the local machine, bound to localhost using the default Redis port. It probably wouldn't make a lot of sense to use a remote service as a persistence store since its primary purpose is to protect from unreliable network connections. Thus, using a local service seems the proper choice. But it would be trivial to update the RedisClient constructor RedisPersistence::new() to take a URI string to specify another server or port.

The MQTT Persistence Model

The Paho Rust library contains a trait that can be used to supply a user-defined persistence:

pub trait ClientPersistence {
	fn open(&mut self, client_id: &str, server_uri: &str) -> MqttResult<()>;
	fn close(&mut self) -> MqttResult<()>;

	fn put(&mut self, key: &str, buffers: Vec<&[u8]>) -> MqttResult<()>;
	fn get(&self, key: &str) -> MqttResult<Vec<u8>>;

	fn remove(&mut self, key: &str) -> MqttResult<()>;
	fn keys(&self) -> MqttResult<Vec<String>>;
	fn clear(&mut self) -> MqttResult<()>; 
	fn contains_key(&self, key: &str) -> bool;
}

These operations map closely to those provided by key/value stores such as hash maps. Redis can implement these operations on a one-to-one basis using a hash type, with the operations hset, hget, hdel, hkeys, del, and hexists. The key string sent to the functions can act as the Redis hash field.

The name (primary key) for the hash is generated by concatenating the client_id and server_uri as provided to the call to open().

The bulk of this library is dedicated to the implementation of a RedisPersistence struct which implements the ClientPersistence trait for use with a Redis server.

The MQTT Rust Client

Using the Redis persisence is fairly trivial. There's an example application, redis_persist_pub.rs demonstrating its use. Simply do the following:

1. Create an instance of a RedisPersistencestruct.
2. Create an instance of an MQTT CreateOptions struct, specifying the RedisPersistence object as the user-defined persistence.
3. Create an MQTT client, using the options.

It can be done like this:

let persistence = RedisPersistence::new();

let create_opts = mqtt::CreateOptionsBuilder::new()
                     .server_uri("tcp://localhost:1883")
                     .user_persistence(persistence)
                     .finalize();

let cli = mqtt::AsyncClient::new(create_opts).unwrap_or_else(|e| {
    println!("Error creating the client: {:?}", e);
    process::exit(1);
});

The library will then automatically use the Redis persistence to save and restore messages and other data as needed.