Telemetry JobRuns.

1 files changed, 33 insertions, 11 deletions

diff --git a/Software/Visual_Studio/Tango.Telemetry/TelemetryPublisher.cs b/Software/Visual_Studio/Tango.Telemetry/TelemetryPublisher.cs
index 42121d884..4c91e6733 100644
--- a/Software/Visual_Studio/Tango.Telemetry/TelemetryPublisher.cs
+++ b/Software/Visual_Studio/Tango.Telemetry/TelemetryPublisher.cs
@@ -194,8 +194,8 @@ namespace Tango.Telemetry
             var source = sender as ITelemetrySource;
             if (source != null)
             {
-                LogManager.Log($"Telemetry stream received {source.Name} -> {e.SourceType} -> {e.TelemetryObject.ToTelemetryName()}.", LogCategory.Debug);
-                PushTelemetryPackage(source, e.TelemetryObject, e.SourceType);
+                LogManager.Log($"Telemetry stream received {source.Name} -> {e.TelemetryObject.ToTelemetryName()}.", LogCategory.Debug);
+                PushTelemetryPackage(source, e.TelemetryObject, TelemetrySourceTypes.Streaming);
             }
         }
 
@@ -408,29 +408,45 @@ namespace Tango.Telemetry
                     {
                         try
                         {
-                            TelemetryHistorySourceCheckPoint checkPoint = StorageManager.GetHistorySourceCheckPoint(source);
+                            TelemetryHistorySourceCheckPoint checkpoint = StorageManager.GetHistorySourceCheckPoint(source);
 
-                            LogManager.Log($"Evaluating history source '{source.Name}' at checkpoint time {checkPoint?.Time:u}", LogCategory.Debug);
+                            if (checkpoint == null)
+                            {
+                                checkpoint = new TelemetryHistorySourceCheckPoint();
+                                checkpoint.SourceName = source.Name;
+                                checkpoint.Time = source.Direction == TelemetryHistorySourceDirection.Ascending ? DateTime.MinValue : DateTime.MaxValue;
+                            }
 
-                            if (await source.CanRequestHistory(checkPoint.Time))
+                            LogManager.Log($"Evaluating history source '{source.Name}' at checkpoint time {checkpoint?.Time:u}", LogCategory.Debug);
+
+                            if (await source.CanRequestHistory(checkpoint.Time))
                             {
-                                var historyTelemetries = (await source.RequestHistory(checkPoint.Time)).OrderBy(x => x.Time).ToList();
+                                List<ITelemetry> historyTelemetries = new List<ITelemetry>();
+
+                                if (source.Direction == TelemetryHistorySourceDirection.Ascending)
+                                {
+                                    historyTelemetries = (await source.RequestHistory(checkpoint.Time)).OrderBy(x => x.Time).ToList();
+                                }
+                                else
+                                {
+                                    historyTelemetries = (await source.RequestHistory(checkpoint.Time)).OrderByDescending(x => x.Time).ToList();
+                                }
 
                                 LogManager.Log($"History source '{source.Name}' returned {historyTelemetries.Count} telemetry items.", LogCategory.Debug);
 
                                 foreach (var telemetry in historyTelemetries)
                                 {
                                     await PushTelemetryPackageAwait(source, telemetry, TelemetrySourceTypes.ExternalStorage);
-                                    checkPoint.Time = telemetry.Time;
-                                    checkPoint.TotalCount++;
-                                    StorageManager.SetHistorySourceCheckPoint(source, checkPoint.Time, checkPoint.TotalCount);
+                                    checkpoint.Time = telemetry.Time;
+                                    checkpoint.TotalCount++;
+                                    StorageManager.SetHistorySourceCheckPoint(source, checkpoint.Time, checkpoint.TotalCount);
                                 }
 
-                                LogManager.Log($"Checkpoint updated for source '{source.Name}': time = {checkPoint.Time:u}, total = {checkPoint.TotalCount}", LogCategory.Debug);
+                                LogManager.Log($"Checkpoint updated for source '{source.Name}': time = {checkpoint.Time:u}, total = {checkpoint.TotalCount}", LogCategory.Debug);
                             }
                             else
                             {
-                                LogManager.Log($"History request for source '{source.Name}' was not permitted at checkpoint time {checkPoint?.Time:u}", LogCategory.Debug);
+                                LogManager.Log($"History request for source '{source.Name}' was not permitted at checkpoint time {checkpoint?.Time:u}", LogCategory.Debug);
                             }
                         }
                         catch (Exception ex)
@@ -610,17 +626,23 @@ namespace Tango.Telemetry
 
             // Prepare standard metadata properties attached to all telemetry sent
             var telemetryName = package.PendingTelemetry.TelemetryObject.ToTelemetryName();
+            var telemetryVersion = package.PendingTelemetry.TelemetryObject.ToTelemetryVersion();
 
             List<KeyValuePair<String, String>> properties = new List<KeyValuePair<string, string>>();
             properties.Add(new KeyValuePair<string, string>("SerialNumber", Config.SerialNumber));
             properties.Add(new KeyValuePair<string, string>("MachineType", Config.MachineType.ToShortName()));
+            properties.Add(new KeyValuePair<string, string>("Organization", Config.Organization));
+            properties.Add(new KeyValuePair<string, string>("Site", Config.Site));
             properties.Add(new KeyValuePair<string, string>("Environment", Config.Environment));
             properties.Add(new KeyValuePair<string, string>("Type", telemetryName));
 
             //Setting telemetry package basic properties for destination..
             package.TelemetryName = telemetryName;
+            package.TelemetryVersion = telemetryVersion;
             package.SerialNumber = Config.SerialNumber;
             package.Environment = Config.Environment;
+            package.Organization = Config.Organization;
+            package.Site = Config.Site;
             package.MachineType = Config.MachineType.ToShortName();