amikhalev/sprinklers_rs
0
1
Fork 0
Code Issues 2 Pull Requests Releases Wiki Activity

Shorted MQTT publishing code
All checks were successful
continuous-integration/drone/push Build is passing
Details

Browse Source
This commit is contained in:
Alex Mikhalev 2020-09-27 22:20:15 -06:00
parent cc88083b19
commit 742a00892d
1 changed files with 32 additions and 66 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

98
src/mqtt_interface.rs
View File

@ -125,34 +125,34 @@ impl MqttInterface {
(Self { client, topics }, event_loop) (Self { client, topics }, event_loop)
} }
async fn publish_connected(&mut self, connected: bool) -> eyre::Result<()> { async fn publish_data<P>(&mut self, topic: String, payload: &P) -> eyre::Result<()>
where
P: serde::Serialize,
{
let payload_vec =
serde_json::to_vec(payload).wrap_err("failed to serialize publish payload")?;
self.client self.client
.publish( .publish(topic, QoS::AtLeastOnce, true, payload_vec)
self.topics.connected(),
QoS::AtLeastOnce,
true,
connected.to_string(),
)
.await .await
.wrap_err("failed to publish connected topic")?; .wrap_err("failed to publish")?;
Ok(()) Ok(())
} }
async fn publish_connected(&mut self, connected: bool) -> eyre::Result<()> {
self.publish_data(self.topics.connected(), &connected)
.await
.wrap_err("failed to publish connected topic")
}
async fn cancel(&mut self) -> Result<(), rumqttc::ClientError> { async fn cancel(&mut self) -> Result<(), rumqttc::ClientError> {
self.client.cancel().await self.client.cancel().await
} }
pub async fn publish_sections(&mut self, sections: &Sections) -> eyre::Result<()> { pub async fn publish_sections(&mut self, sections: &Sections) -> eyre::Result<()> {
let section_ids: Vec<_> = sections.keys().cloned().collect(); let section_ids: Vec<_> = sections.keys().cloned().collect();
let section_ids_payload = serde_json::to_vec(&section_ids)?; self.publish_data(self.topics.sections(), &section_ids)
self.client .await
.publish( .wrap_err("failed to publish section ids")?;
self.topics.sections(),
QoS::AtLeastOnce,
true,
section_ids_payload,
)
.await?;
for section in sections.values() { for section in sections.values() {
self.publish_section(section).await?; self.publish_section(section).await?;
} }
@ -160,16 +160,9 @@ impl MqttInterface {
} }
pub async fn publish_section(&mut self, section: &Section) -> eyre::Result<()> { pub async fn publish_section(&mut self, section: &Section) -> eyre::Result<()> {
let payload = serde_json::to_vec(section).wrap_err("failed to serialize section")?; self.publish_data(self.topics.section_data(section.id), section)
self.client .await
.publish( .wrap_err("failed to publish section")
self.topics.section_data(section.id),
QoS::AtLeastOnce,
true,
payload,
)
.await?;
Ok(())
} }
pub async fn publish_section_state( pub async fn publish_section_state(
@ -177,29 +170,16 @@ impl MqttInterface {
section_id: SectionId, section_id: SectionId,
state: bool, state: bool,
) -> eyre::Result<()> { ) -> eyre::Result<()> {
let payload: Vec<u8> = state.to_string().into(); self.publish_data(self.topics.section_state(section_id), &state)
self.client .await
.publish( .wrap_err("failed to publish section state")
self.topics.section_state(section_id),
QoS::AtLeastOnce,
true,
payload,
)
.await?;
Ok(())
} }
pub async fn publish_programs(&mut self, programs: &Programs) -> eyre::Result<()> { pub async fn publish_programs(&mut self, programs: &Programs) -> eyre::Result<()> {
let program_ids: Vec<_> = programs.keys().cloned().collect(); let program_ids: Vec<_> = programs.keys().cloned().collect();
let program_ids_payload = serde_json::to_vec(&program_ids)?; self.publish_data(self.topics.programs(), &program_ids)
self.client .await
.publish( .wrap_err("failed to publish program ids")?;
self.topics.programs(),
QoS::AtLeastOnce,
true,
program_ids_payload,
)
.await?;
for program in programs.values() { for program in programs.values() {
self.publish_program(program).await?; self.publish_program(program).await?;
} }
@ -207,16 +187,9 @@ impl MqttInterface {
} }
pub async fn publish_program(&mut self, program: &Program) -> eyre::Result<()> { pub async fn publish_program(&mut self, program: &Program) -> eyre::Result<()> {
let payload = serde_json::to_vec(program).wrap_err("failed to serialize program")?; self.publish_data(self.topics.program_data(program.id), &program)
self.client .await
.publish( .wrap_err("failed to publish program")
self.topics.program_data(program.id),
QoS::AtLeastOnce,
true,
payload,
)
.await?;
Ok(())
} }
pub async fn publish_program_running( pub async fn publish_program_running(
@ -224,16 +197,9 @@ impl MqttInterface {
program_id: ProgramId, program_id: ProgramId,
running: bool, running: bool,
) -> eyre::Result<()> { ) -> eyre::Result<()> {
let payload = running.to_string(); self.publish_data(self.topics.program_running(program_id), &running)
self.client .await
.publish( .wrap_err("failed to publish program running")
self.topics.program_running(program_id),
QoS::AtLeastOnce,
true,
payload,
)
.await?;
Ok(())
} }
} }