Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 11 additions & 11 deletions src/SparkplugNet/VersionB/SparkplugApplication.cs
Original file line number Diff line number Diff line change
Expand Up @@ -149,32 +149,32 @@ private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payloa
{
// Filter out session number metric.
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name == Constants.SessionNumberMetricName);
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
var metrics = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName).ToList();
// var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();

if (sessionNumberMetric is not null)
{
filteredMetrics.Add(sessionNumberMetric);
metrics.Add(sessionNumberMetric);
}

// Handle messages.
switch (topic.MessageType)
{
case SparkplugMessageType.NodeBirth:
await this.FireNodeBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier,
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Online));
break;
case SparkplugMessageType.DeviceBirth:
if (string.IsNullOrWhiteSpace(topic.DeviceIdentifier))
{
throw new InvalidOperationException($"The device identifier is invalid!");
}

await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier,
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Online));
break;
case SparkplugMessageType.NodeData:
var nodeDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
var nodeDataMetrics = this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Online);
await this.FireNodeDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, nodeDataMetrics);
break;
case SparkplugMessageType.DeviceData:
Expand All @@ -183,20 +183,20 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
throw new InvalidOperationException($"Topic {topic} is invalid!");
}

var deviceDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
var deviceDataMetrics = this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Online);
await this.FireDeviceDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier, deviceDataMetrics);
break;
case SparkplugMessageType.NodeDeath:
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Offline);
await this.FireNodeDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, sessionNumberMetric);
break;
break;
case SparkplugMessageType.DeviceDeath:
if (string.IsNullOrWhiteSpace(topic.DeviceIdentifier))
{
throw new InvalidOperationException($"Topic {topic} is invalid!");
}

this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
this.ProcessPayload(topic, metrics, SparkplugMetricStatus.Offline);
await this.FireDeviceDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier);
break;
}
Expand Down
10 changes: 5 additions & 5 deletions src/SparkplugNet/VersionB/SparkplugNode.cs
Original file line number Diff line number Diff line change
Expand Up @@ -114,12 +114,12 @@ private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payloa
{
// Filter out session number metric.
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name == Constants.SessionNumberMetricName);
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
var metrics = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName).ToList();
// var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();

if (sessionNumberMetric is not null)
{
filteredMetrics.Add(sessionNumberMetric);
metrics.Add(sessionNumberMetric);
}

// Handle messages.
Expand All @@ -131,11 +131,11 @@ private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payloa
throw new InvalidOperationException($"Topic {topic} is invalid!");
}

await this.FireDeviceCommandReceived(topic.DeviceIdentifier, filteredMetrics);
await this.FireDeviceCommandReceived(topic.DeviceIdentifier, metrics);
break;

case SparkplugMessageType.NodeCommand:
await this.FireNodeCommandReceived(filteredMetrics);
await this.FireNodeCommandReceived(metrics);
break;
}
}
Expand Down