Merge pull request #59 from osmanhadzic/astarte-stored-individual-mes…

…sage Implement fallout strategy failed individual message
astarte-platform · Feb 5, 2024 · 396d7a0 · 396d7a0
2 parents 5cc882b + 482d53e
commit 396d7a0
Show file tree

Hide file tree

Showing 2 changed files with 15 additions and 3 deletions.
diff --git a/AstarteDeviceSDKCSharp/Transport/MQTT/AstarteMqttV1Transport.cs b/AstarteDeviceSDKCSharp/Transport/MQTT/AstarteMqttV1Transport.cs
@@ -46,7 +46,7 @@ public AstarteMqttV1Transport(MutualSSLAuthenticationMqttConnectionInfo connecti
         public override async Task SendIndividualValue(AstarteInterface astarteInterface,
         string path, object? value, DateTime? timestamp)
         {
-            AstarteInterfaceDatastreamMapping mapping;
+            AstarteInterfaceDatastreamMapping mapping = new();
             int qos = 2;
 
             if (astarteInterface.GetType() == (typeof(AstarteDeviceDatastreamInterface)))
@@ -67,7 +67,15 @@ public override async Task SendIndividualValue(AstarteInterface astarteInterface
             string topic = _baseTopic + "/" + astarteInterface.InterfaceName + path;
             byte[] payload = AstartePayload.Serialize(value, timestamp);
 
-            await DoSendMqttMessage(topic, payload, qos);
+            try
+            {
+                await DoSendMqttMessage(topic, payload, qos);
+            }
+            catch (MqttCommunicationException ex)
+            {
+                HandleDatastreamFailedPublish(ex, mapping, topic, payload, qos);
+            }
+
         }
 
         private async Task DoSendMqttMessage(string topic, byte[] payload, int qos)
@@ -94,6 +102,7 @@ private async Task DoSendMqttMessage(string topic, byte[] payload, int qos)
             catch (Exception)
             {
                 _astarteTransportEventListener?.OnTransportDisconnected();
+                throw new MqttCommunicationException(topic);
             }
 
         }
@@ -156,7 +165,6 @@ public override async Task SendAggregate(AstarteAggregateDatastreamInterface ast
             }
             catch (MqttCommunicationException e)
             {
-
                 HandleDatastreamFailedPublish(e, mapping, topic, payload, qos);
             }
         }

diff --git a/CHANGELOG.md b/CHANGELOG.md
@@ -4,6 +4,10 @@ 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.7.0] - Unreleased
+### Added
+- Add a fallout strategy for individual failed messages.
+
 ## [0.6.0] - 2023-12-18
 ### Added
 - Add the capability to update the introspection dynamically.