The actix-mqtt-client
crate is a mqtt 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: ```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(),
);
log::info!("Connect");
client.connect().map(|_| client)
})
.and_then(|client| {
log::info!("Subscribe");
client
.subscribe(String::from("topic"), mqtt::QualityOfService::Level2)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level0");
client
.publish(
String::from("topic"),
mqtt::QualityOfService::Level0,
Vec::from("level0".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level1");
client
.publish(
String::from("topic"),
mqtt::QualityOfService::Level1,
Vec::from("level1".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Publish Level2");
client
.publish(
String::from("topic"),
mqtt::QualityOfService::Level2,
Vec::from("level2".as_bytes()),
)
.map(|_| client)
})
.and_then(|client| {
log::info!("Disconnect");
client.disconnect()
})
.map_err(|_| ()),
);
})
.unwrap();