The actix-mqtt-client
crate is a mqtt v3.1.1 client based on the actix framework
First, create 2 actix actors, one for receiving publish messages, the other one for receiving error messages from the client, you can also create an optional actix actor for receiving the stop message: ```rust pub struct ErrorActor;
impl actix::Actor for ErrorActor {
type Context = actix::Context
impl actix::Handler
pub struct MessageActor;
impl actix::Actor for MessageActor {
type Context = actix::Context
impl actix::Handler
Then, connect to the server(using tokio) and use the read and write part of the stream along with the actors to create a MqttClient:
rust
System::run(|| {
let socket_addr = SocketAddr::from_str("127.0.0.1:1883").unwrap();
Arbiter::spawn(
TcpStream::connect(&socket_addr)
.and_then(|stream| {
let (r, w) = stream.split();
let mut client = MqttClient::new(
r,
w,
String::from("mqtt_client"),
MqttOptions::default(),
MessageActor.start().recipient(),
ErrorActor.start().recipient(),
None
);
log::info!("Connect");
client.connect().map(|_| client)
})
.and_then(|client| {
log::info!("Subscribe");
client
.subscribe(String::from("topic"), QualityOfService::Level2)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level0");
client
.publish(
String::from("topic"),
QualityOfService::Level0,
Vec::from("level0".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level1");
client
.publish(
String::from("topic"),
QualityOfService::Level1,
Vec::from("level1".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level2");
client
.publish(
String::from("topic"),
QualityOfService::Level2,
Vec::from("level2".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Disconnect");
client.disconnect(false)
})
.map_err(|_| ()),
);
})
.unwrap();