fix: QoS conversion

risingwavelabs · Mar 5, 2024 · fb3daee · fb3daee
1 parent b8c2fc7
commit fb3daee
Show file tree

Hide file tree

Showing 3 changed files with 6 additions and 6 deletions.
diff --git a/src/connector/src/sink/mqtt.rs b/src/connector/src/sink/mqtt.rs
@@ -178,8 +178,8 @@ impl MqttSinkWriter {
             .qos
             .as_ref()
             .map(|qos| match qos {
-                QualityOfService::AtLeastOnce => QoS::AtMostOnce,
-                QualityOfService::AtMostOnce => QoS::AtLeastOnce,
+                QualityOfService::AtMostOnce => QoS::AtMostOnce,
+                QualityOfService::AtLeastOnce => QoS::AtLeastOnce,
                 QualityOfService::ExactlyOnce => QoS::ExactlyOnce,
             })
             .unwrap_or(QoS::AtMostOnce);

diff --git a/src/connector/src/source/mqtt/enumerator/mod.rs b/src/connector/src/source/mqtt/enumerator/mod.rs
@@ -54,7 +54,7 @@ impl SplitEnumerator for MqttSplitEnumerator {
         }
 
         client
-            .subscribe(topic.clone(), rumqttc::v5::mqttbytes::QoS::AtLeastOnce)
+            .subscribe(topic.clone(), rumqttc::v5::mqttbytes::QoS::AtMostOnce)
             .await?;
 
         let cloned_client = client.clone();
@@ -98,7 +98,7 @@ impl SplitEnumerator for MqttSplitEnumerator {
                         );
                         connected_clone.store(false, std::sync::atomic::Ordering::Relaxed);
                         cloned_client
-                            .subscribe(topic.clone(), rumqttc::v5::mqttbytes::QoS::AtLeastOnce)
+                            .subscribe(topic.clone(), rumqttc::v5::mqttbytes::QoS::AtMostOnce)
                             .await
                             .unwrap();
                     }

diff --git a/src/connector/src/source/mqtt/source/reader.rs b/src/connector/src/source/mqtt/source/reader.rs
@@ -58,8 +58,8 @@ impl SplitReader for MqttSplitReader {
             .qos
             .as_ref()
             .map(|qos| match qos {
-                QualityOfService::AtLeastOnce => QoS::AtMostOnce,
-                QualityOfService::AtMostOnce => QoS::AtLeastOnce,
+                QualityOfService::AtMostOnce => QoS::AtMostOnce,
+                QualityOfService::AtLeastOnce => QoS::AtLeastOnce,
                 QualityOfService::ExactlyOnce => QoS::ExactlyOnce,
             })
             .unwrap_or(QoS::AtMostOnce);