Implement async StreamData method

Refactor method for `StreamData` for individual and aggregate data to `Astarte` Signed-off-by: Osman Hadzic <osman.hadzic@secomind.com>
astarte-platform · Apr 24, 2024 · 2ae33dc · 2ae33dc
1 parent 1ac497d
commit 2ae33dc
Show file tree

Hide file tree

Showing 6 changed files with 15 additions and 14 deletions.
diff --git a/AstarteDeviceSDKCSharp/Protocol/AstarteDeviceAggregateDatastreamInterface.cs b/AstarteDeviceSDKCSharp/Protocol/AstarteDeviceAggregateDatastreamInterface.cs
@@ -27,12 +27,12 @@ namespace AstarteDeviceSDKCSharp.Protocol
 {
     public class AstarteDeviceAggregateDatastreamInterface : AstarteAggregateDatastreamInterface, IAstarteAggregateDataStreamer
     {
-        public void StreamData(string path, Dictionary<string, object> payload)
+        public async Task StreamData(string path, Dictionary<string, object> payload)
         {
-            StreamData(path, payload, null);
+            await StreamData(path, payload, null);
         }
 
-        public void StreamData(string path, Dictionary<string, object> payload, DateTime? timestamp)
+        public async Task StreamData(string path, Dictionary<string, object> payload, DateTime? timestamp)
         {
             ValidatePayload(path, payload, timestamp);
 
@@ -42,7 +42,7 @@ public void StreamData(string path, Dictionary<string, object> payload, DateTime
                 throw new AstarteTransportException("No available transport");
             }
 
-            transport.SendAggregate(this, path, payload, timestamp);
+            await transport.SendAggregate(this, path, payload, timestamp);
         }
 
         public void ValidatePayload(string path, Dictionary<string, object> payload, DateTime? timestamp)

diff --git a/AstarteDeviceSDKCSharp/Protocol/AstarteDeviceDatastreamInterface.cs b/AstarteDeviceSDKCSharp/Protocol/AstarteDeviceDatastreamInterface.cs
@@ -24,12 +24,12 @@ namespace AstarteDeviceSDKCSharp.Protocol
 {
     public class AstarteDeviceDatastreamInterface : AstarteDatastreamInterface, IAstarteDataStreamer
     {
-        public void StreamData(string path, object payload)
+        public async Task StreamData(string path, object payload)
         {
-            StreamData(path, payload, null);
+            await StreamData(path, payload, null);
         }
 
-        public void StreamData(string path, object payload, DateTime? timestamp)
+        public async Task StreamData(string path, object payload, DateTime? timestamp)
         {
             ValidatePayload(path, payload, timestamp);
             AstarteTransport? transport = GetAstarteTransport();
@@ -38,7 +38,7 @@ public void StreamData(string path, object payload, DateTime? timestamp)
                 throw new AstarteTransportException("No available transport");
             }
 
-            transport.SendIndividualValue(this, path, payload, timestamp);
+            await transport.SendIndividualValue(this, path, payload, timestamp);
         }
     }
 }
diff --git a/AstarteDeviceSDKCSharp/Protocol/IAstarteAggregateDataStreamer.cs b/AstarteDeviceSDKCSharp/Protocol/IAstarteAggregateDataStreamer.cs
@@ -22,8 +22,8 @@ namespace AstarteDeviceSDKCSharp.Protocol
 {
     public interface IAstarteAggregateDataStreamer
     {
-        void StreamData(String path, Dictionary<String, Object> payload);
+        Task StreamData(String path, Dictionary<String, Object> payload);
 
-        void StreamData(String path, Dictionary<String, Object> payload, DateTime? timestamp);
+        Task StreamData(String path, Dictionary<String, Object> payload, DateTime? timestamp);
     }
 }
diff --git a/AstarteDeviceSDKCSharp/Protocol/IAstarteDataStreamer.cs b/AstarteDeviceSDKCSharp/Protocol/IAstarteDataStreamer.cs
@@ -27,14 +27,14 @@ public interface IAstarteDataStreamer
         /// </summary>
         /// <param name="path">Endpoint</param>
         /// <param name="payload">Message for MQTT broker</param>
-        void StreamData(String path, Object payload);
+        Task StreamData(String path, Object payload);
 
         /// <summary>
         /// Method for sending individual values to Astarte with timestamp
         /// </summary>
         /// <param name="path">Endpoint</param>
         /// <param name="payload">Message for MQTT broker</param>
         /// <param name="timestamp">UTC</param>
-        void StreamData(String path, Object payload, DateTime? timestamp);
+        Task StreamData(String path, Object payload, DateTime? timestamp);
     }
 }
diff --git a/AstarteDeviceSDKCSharpE2E.Tests/AstarteAggregateDeviceTest.cs b/AstarteDeviceSDKCSharpE2E.Tests/AstarteAggregateDeviceTest.cs
@@ -73,7 +73,7 @@ public async Task AggregateFromDeviceToServer()
             (AstarteDeviceAggregateDatastreamInterface)astarteDevice
             .GetInterface(interfaceName);
 
-            astarteDeviceAggregateDatastream
+            await astarteDeviceAggregateDatastream
             .StreamData("/%{sensor_id}", astarteMockDevice.MockDataDictionary, DateTime.Now);
             Thread.Sleep(500);
 

diff --git a/AstarteDeviceSDKCSharpE2E.Tests/AstarteDatastreamDeviceTests.cs b/AstarteDeviceSDKCSharpE2E.Tests/AstarteDatastreamDeviceTests.cs
@@ -78,7 +78,8 @@ public async Task DatastreamFromDeviceToServer()
 
             foreach (var item in astarteMockDevice.MockDataDictionary)
             {
-                astarteDatastreamInterface.StreamData($"/{item.Key}", item.Value, DateTime.Now);
+                await astarteDatastreamInterface
+                .StreamData($"/{item.Key}", item.Value, DateTime.Now);
                 Thread.Sleep(500);
             }