release 0.4.0

foxzool · foxzool · commit 559d85f8f68d · 2024-09-26T16:15:28.000+08:00
diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -5,6 +5,11 @@ All notable changes to this project will be documented in this file.
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.0.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
 
+## [0.4.0] - 2024-09-26
+
+- every MQTT client is a component now
+- can spawn a `SubscribeTopic` component to subscribe to a topic and observe the messages
+
 ## [0.3.3] - 2024-09-24
 
 - Update MQTT example and library handling: Refactor error handling, subscribe method, and event types for clarity and
diff --git a/Cargo.toml b/Cargo.toml
@@ -1,7 +1,7 @@
 [package]
 name = "bevy_mqtt"
 description = "A simple MQTT client for Bevy ECS"
-version = "0.3.3"
+version = "0.4.0"
 edition = "2021"
 readme = "README.md"
 repository = "https://github.com/foxzool/bevy_mqtt"
@@ -21,7 +21,6 @@ bevy_app = { version = "0.14.2" }
 bevy_ecs = { version = "0.14.2" }
 bevy_log = { version = "0.14.2" }
 bevy_hierarchy = { version = "0.14.2" }
-bevy_derive = { version = "0.14.2" }
 
 rumqttc = { version = "0.24.0" }
 flume = { version = "0.11.0" }
diff --git a/README.md b/README.md
@@ -15,71 +15,85 @@ first run as mqtt broker like [mosquitto](https://mosquitto.org/)
 then run the example
 
 ```rust
-use std::time::SystemTime;
-
 use bevy::{prelude::*, time::common_conditions::on_timer};
 use bevy_log::LogPlugin;
 use bevy_mqtt::{
-    rumqttc::{MqttOptions, QoS},
-    MqttClient, MqttClientError, MqttClientState, MqttConnectError, MqttEvent, MqttPlugin,
-    MqttPublishOutgoing, MqttSetting, SubscribeTopic, TopicMessage,
+    rumqttc::QoS, MqttClient, MqttClientConnected, MqttClientError, MqttConnectError, MqttEvent,
+    MqttPlugin, MqttSetting, SubscribeTopic, TopicMessage,
 };
-use bevy_state::prelude::OnEnter;
-use bincode::ErrorKind;
-use serde::{Deserialize, Serialize};
-
-#[derive(Serialize, Deserialize, Debug)]
-struct Message {
-    i: usize,
-    time: SystemTime,
-}
-
-impl From<&Message> for Vec<u8> {
-    fn from(value: &Message) -> Self {
-        bincode::serialize(value).unwrap()
-    }
-}
-
-impl From<Message> for Vec<u8> {
-    fn from(value: Message) -> Self {
-        bincode::serialize(&value).unwrap()
-    }
-}
-
-impl TryFrom<&[u8]> for Message {
-    type Error = Box<ErrorKind>;
-
-    fn try_from(value: &[u8]) -> Result<Self, Self::Error> {
-        bincode::deserialize(value)
-    }
-}
+use rumqttc::{MqttOptions, Transport};
+use std::time::Duration;
 
 fn main() {
     App::new()
-        .insert_resource(MqttSetting {
-            mqtt_options: MqttOptions::new("mqtt-serde", "127.0.0.1", 1883),
-            cap: 10,
-        })
         .add_plugins((MinimalPlugins, MqttPlugin, LogPlugin::default()))
-        .add_systems(Update, (handle_message, handle_error))
-        .add_systems(
-            OnEnter(MqttClientState::Connected),
-            (sub_topic_direct, sub_topic_by_component),
-        )
+        .add_systems(Startup, setup_clients)
+        .add_systems(Update, (sub_topic, handle_message, handle_error))
         .add_systems(
             Update,
             publish_message.run_if(on_timer(std::time::Duration::from_secs(1))),
         )
         .run();
 }
 
+fn setup_clients(mut commands: Commands) {
+    commands.spawn(MqttSetting {
+        mqtt_options: MqttOptions::new("bevy-mqtt-client", "127.0.0.1", 1883),
+        cap: 10,
+    });
+
+    let mut mqtt_options = MqttOptions::new("mqtt-ws-client", "ws://127.0.0.1:8080", 8080);
+    mqtt_options.set_transport(Transport::Ws);
+    // mqtt_options.set_credentials("username", "password");
+    mqtt_options.set_keep_alive(Duration::from_secs(5));
+
+    commands.spawn((
+        MqttSetting {
+            mqtt_options,
+            cap: 10,
+        },
+        WebsocketMqttClient,
+    ));
+}
+
+#[derive(Component)]
+struct WebsocketMqttClient;
+
+/// this is a system that subscribes to a topic and handle the incoming messages
+fn sub_topic(
+    mqtt_client: Query<(Entity, &MqttClient, &MqttSetting), Added<MqttClientConnected>>,
+    mut commands: Commands,
+) {
+    for (entity, client, setting) in mqtt_client.iter() {
+        client
+            .subscribe("hello".to_string(), QoS::AtMostOnce)
+            .unwrap();
+
+        let setting = setting.clone();
+        let child_id = commands
+            .spawn(SubscribeTopic::new("+/mqtt", QoS::AtMostOnce))
+            .observe(move |topic_message: Trigger<TopicMessage>| {
+                println!(
+                    "{:?}: Topic: '+/mqtt' received : {:?}",
+                    setting.mqtt_options.broker_address().clone(),
+                    topic_message.event().payload
+                );
+            })
+            .id();
+        commands.entity(entity).add_child(child_id);
+    }
+}
+
+/// this is global handler for all incoming messages
 fn handle_message(mut mqtt_event: EventReader<MqttEvent>) {
     for event in mqtt_event.read() {
-        match &event.0 {
+        match &event.event {
             rumqttc::Event::Incoming(income) => match income {
                 rumqttc::Incoming::Publish(publish) => {
-                    let message: Message = bincode::deserialize(&publish.payload).unwrap();
-                    println!("Received Publish: {:?}", message);
+                    println!(
+                        "Topic Component: {} Received: {:?}",
+                        publish.topic, publish.payload
+                    );
                 }
                 _ => {
                     println!("Incoming: {:?}", income);
@@ -103,56 +117,25 @@ fn handle_error(
     }
 }
 
-/// there are two ways to subscribe to a topic
-/// 1. Directly subscribe to a topic
-/// 2. Subscribe to a topic by component
-fn sub_topic_direct(client: Res<MqttClient>) {
-    client
-        .try_subscribe("hello/mqtt", QoS::AtMostOnce)
-        .expect("subscribe failed");
-}
-
-fn sub_topic_by_component(mut commands: Commands) {
-    commands
-        .spawn(SubscribeTopic::new("+/mqtt", QoS::AtMostOnce))
-        .observe(|topic_message: Trigger<TopicMessage>| {
-            println!(
-                "topic: {} received : {:?}",
-                topic_message.event().topic,
-                topic_message.event().payload
-            );
-        });
-}
-
-fn publish_message(mut pub_events: EventWriter<MqttPublishOutgoing>) {
-    let mut list = vec![];
-    for i in 0..3 {
-        let message = Message {
-            i,
-            time: SystemTime::now(),
-        };
-
-        list.push(MqttPublishOutgoing {
-            topic: "hello/mqtt".to_string(),
-            qos: QoS::AtLeastOnce,
-            retain: false,
-            payload: message.into(),
-        });
-        list.push(MqttPublishOutgoing {
-            topic: "bevy/mqtt".to_string(),
-            qos: QoS::AtLeastOnce,
-            retain: false,
-            payload: Message {
-                i: 999,
-                time: SystemTime::now(),
-            }
-                .into(),
-        });
+fn publish_message(mqtt_client: Query<&MqttClient, With<MqttClientConnected>>) {
+    for client in mqtt_client.iter() {
+        client
+            .publish(
+                "hello".to_string(),
+                QoS::AtMostOnce,
+                false,
+                "mqtt".as_bytes(),
+            )
+            .unwrap();
+        for i in 0..3 {
+            client
+                .publish(format!("{}/mqtt", i), QoS::AtMostOnce, false, b"hello")
+                .unwrap();
+        }
     }
-
-    pub_events.send_batch(list);
 }
 
+
 ```
 
 ## Supported Versions
diff --git a/examples/pub_and_sub.rs b/examples/pub_and_sub.rs
@@ -44,19 +44,21 @@ struct WebsocketMqttClient;
 
 /// this is a system that subscribes to a topic and handle the incoming messages
 fn sub_topic(
-    mqtt_client: Query<(Entity, &MqttClient), Added<MqttClientConnected>>,
+    mqtt_client: Query<(Entity, &MqttClient, &MqttSetting), Added<MqttClientConnected>>,
     mut commands: Commands,
 ) {
-    for (entity, client) in mqtt_client.iter() {
+    for (entity, client, setting) in mqtt_client.iter() {
         client
             .subscribe("hello".to_string(), QoS::AtMostOnce)
             .unwrap();
 
+        let setting = setting.clone();
         let child_id = commands
             .spawn(SubscribeTopic::new("+/mqtt", QoS::AtMostOnce))
-            .observe(|topic_message: Trigger<TopicMessage>| {
+            .observe(move |topic_message: Trigger<TopicMessage>| {
                 println!(
-                    "Topic: '+/mqtt' received : {:?}",
+                    "{:?}: Topic: '+/mqtt' received : {:?}",
+                    setting.mqtt_options.broker_address().clone(),
                     topic_message.event().payload
                 );
             })
@@ -68,7 +70,7 @@ fn sub_topic(
 /// this is global handler for all incoming messages
 fn handle_message(mut mqtt_event: EventReader<MqttEvent>) {
     for event in mqtt_event.read() {
-        match &event.0 {
+        match &event.event {
             rumqttc::Event::Incoming(income) => match income {
                 rumqttc::Incoming::Publish(publish) => {
                     println!(
diff --git a/src/lib.rs b/src/lib.rs
@@ -1,7 +1,6 @@
 //! A Bevy plugin for MQTT
 
 use bevy_app::{App, Plugin, Update};
-use bevy_derive::{Deref, DerefMut};
 use bevy_ecs::prelude::*;
 use bevy_hierarchy::Parent;
 use bevy_log::{debug, trace};
@@ -73,8 +72,11 @@ impl DerefMut for MqttClient {
 pub struct MqttClientConnected;
 
 /// A wrapper around rumqttc::Event
-#[derive(Debug, Clone, PartialEq, Eq, Deref, DerefMut, Event)]
-pub struct MqttEvent(pub rumqttc::Event);
+#[derive(Debug, Clone, PartialEq, Eq, Event)]
+pub struct MqttEvent {
+    pub entity: Entity,
+    pub event: rumqttc::Event,
+}
 
 /// A wrapper around rumqttc::ConnectionError
 #[derive(Debug, Event)]
@@ -144,7 +146,10 @@ fn handle_mqtt_events(
                 }
                 rumqttc::Event::Incoming(_) | rumqttc::Event::Outgoing(_) => {}
             }
-            mqtt_events.send(MqttEvent(event));
+            mqtt_events.send(MqttEvent {
+                entity,
+                event: event.clone(),
+            });
         }
 
         while let Ok(error) = client.from_async_error.try_recv() {
