From 742a00892d5121583195adda8fcaed9aeb36f9da Mon Sep 17 00:00:00 2001 From: Alex Mikhalev Date: Sun, 27 Sep 2020 22:20:15 -0600 Subject: [PATCH] Shorted MQTT publishing code --- src/mqtt_interface.rs | 98 ++++++++++++++----------------------------- 1 file changed, 32 insertions(+), 66 deletions(-) diff --git a/src/mqtt_interface.rs b/src/mqtt_interface.rs index 81615d9..9026e3f 100644 --- a/src/mqtt_interface.rs +++ b/src/mqtt_interface.rs @@ -125,34 +125,34 @@ impl MqttInterface { (Self { client, topics }, event_loop) } - async fn publish_connected(&mut self, connected: bool) -> eyre::Result<()> { + async fn publish_data

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