Skip to content

Commit 9cb2deb

Browse files
committed
Adjusted filtering of messages. Fixes #48.
1 parent 9ccbddd commit 9cb2deb

File tree

4 files changed

+56
-31
lines changed

4 files changed

+56
-31
lines changed

src/SparkplugNet/VersionA/SparkplugApplication.cs

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -167,22 +167,29 @@ protected override async Task OnMessageReceived(SparkplugMessageTopic topic, byt
167167
/// <exception cref="Exception">Thrown if the metric is unknown.</exception>
168168
private async Task HandleMessagesForVersionA(SparkplugMessageTopic topic, VersionAData.Payload payload)
169169
{
170-
// Filter out settion number metric.
170+
// Filter out session number metric.
171+
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name != Constants.SessionNumberMetricName);
171172
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
172-
this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType);
173+
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
173174

175+
if (sessionNumberMetric is not null)
176+
{
177+
filteredMetrics.Add(sessionNumberMetric);
178+
}
179+
180+
// Handle messages.
174181
switch (topic.MessageType)
175182
{
176183
case SparkplugMessageType.NodeBirth:
177184
await this.FireNodeBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier,
178-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online));
185+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
179186
break;
180187
case SparkplugMessageType.DeviceBirth:
181188
await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.EdgeNodeIdentifier,
182-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online));
189+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
183190
break;
184191
case SparkplugMessageType.NodeData:
185-
var nodeDataMetrics = this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online);
192+
var nodeDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
186193
await this.FireNodeDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, nodeDataMetrics);
187194
break;
188195
case SparkplugMessageType.DeviceData:
@@ -191,11 +198,11 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
191198
throw new InvalidOperationException($"Topic {topic} is invalid!");
192199
}
193200

194-
var deviceDataMetrics = this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online);
201+
var deviceDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
195202
await this.FireDeviceDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier, deviceDataMetrics);
196203
break;
197204
case SparkplugMessageType.NodeDeath:
198-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Offline);
205+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
199206
await this.FireNodeDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier);
200207
break;
201208
case SparkplugMessageType.DeviceDeath:
@@ -204,7 +211,7 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
204211
throw new InvalidOperationException($"Topic {topic} is invalid!");
205212
}
206213

207-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Offline);
214+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
208215
await this.FireDeviceDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier);
209216
break;
210217
}
@@ -214,12 +221,12 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
214221
/// Handles the device message.
215222
/// </summary>
216223
/// <param name="topic">The topic.</param>
217-
/// <param name="payload">The payload.</param>
224+
/// <param name="metrics">The metrics.</param>
218225
/// <param name="metricStatus">The metric status.</param>
219226
/// <exception cref="InvalidCastException">Thrown if the metric cast is invalid.</exception>
220227
private IEnumerable<VersionAData.KuraMetric> ProcessPayload(
221228
SparkplugMessageTopic topic,
222-
VersionAData.Payload payload,
229+
List<VersionAData.KuraMetric> metrics,
223230
SparkplugMetricStatus metricStatus)
224231
{
225232
var metricState = new MetricState<VersionAData.KuraMetric>
@@ -241,7 +248,7 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
241248
this.NodeStates[topic.EdgeNodeIdentifier] = metricState;
242249
}
243250

244-
foreach (var payloadMetric in payload.Metrics)
251+
foreach (var payloadMetric in metrics)
245252
{
246253
if (payloadMetric is not VersionAData.KuraMetric convertedMetric)
247254
{

src/SparkplugNet/VersionA/SparkplugNode.cs

Lines changed: 10 additions & 3 deletions
Original file line numberDiff line numberDiff line change
@@ -119,9 +119,16 @@ protected override async Task OnMessageReceived(SparkplugMessageTopic topic, byt
119119
private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, VersionAData.Payload payload)
120120
{
121121
// Filter out session number metric.
122+
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name != Constants.SessionNumberMetricName);
122123
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
123-
this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType);
124+
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
124125

126+
if (sessionNumberMetric is not null)
127+
{
128+
filteredMetrics.Add(sessionNumberMetric);
129+
}
130+
131+
// Handle messages.
125132
switch (topic.MessageType)
126133
{
127134
case SparkplugMessageType.DeviceCommand:
@@ -130,11 +137,11 @@ private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Versio
130137
throw new InvalidOperationException($"Topic {topic} is invalid!");
131138
}
132139

133-
await this.FireDeviceCommandReceived(topic.DeviceIdentifier, payload.Metrics);
140+
await this.FireDeviceCommandReceived(topic.DeviceIdentifier, filteredMetrics);
134141
break;
135142

136143
case SparkplugMessageType.NodeCommand:
137-
await this.FireNodeCommandReceived(payload.Metrics);
144+
await this.FireNodeCommandReceived(filteredMetrics);
138145
break;
139146
}
140147
}

src/SparkplugNet/VersionB/SparkplugApplication.cs

Lines changed: 18 additions & 11 deletions
Original file line numberDiff line numberDiff line change
@@ -149,15 +149,22 @@ protected override async Task OnMessageReceived(SparkplugMessageTopic topic, byt
149149
/// <exception cref="Exception">Thrown if the metric is unknown.</exception>
150150
private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payload payload)
151151
{
152-
// Filter out settion number metric.
152+
// Filter out session number metric.
153+
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name != Constants.SessionNumberMetricName);
153154
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
154-
this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType);
155+
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
155156

157+
if (sessionNumberMetric is not null)
158+
{
159+
filteredMetrics.Add(sessionNumberMetric);
160+
}
161+
162+
// Handle messages.
156163
switch (topic.MessageType)
157164
{
158165
case SparkplugMessageType.NodeBirth:
159166
await this.FireNodeBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier,
160-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online));
167+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
161168
break;
162169
case SparkplugMessageType.DeviceBirth:
163170
if (string.IsNullOrWhiteSpace(topic.DeviceIdentifier))
@@ -166,10 +173,10 @@ await this.FireNodeBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier
166173
}
167174

168175
await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier,
169-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online));
176+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online));
170177
break;
171178
case SparkplugMessageType.NodeData:
172-
var nodeDataMetrics = this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online);
179+
var nodeDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
173180
await this.FireNodeDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, nodeDataMetrics);
174181
break;
175182
case SparkplugMessageType.DeviceData:
@@ -178,11 +185,11 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
178185
throw new InvalidOperationException($"Topic {topic} is invalid!");
179186
}
180187

181-
var deviceDataMetrics = this.ProcessPayload(topic, payload, SparkplugMetricStatus.Online);
188+
var deviceDataMetrics = this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Online);
182189
await this.FireDeviceDataReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier, deviceDataMetrics);
183190
break;
184191
case SparkplugMessageType.NodeDeath:
185-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Offline);
192+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
186193
await this.FireNodeDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier);
187194
break;
188195
case SparkplugMessageType.DeviceDeath:
@@ -191,7 +198,7 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
191198
throw new InvalidOperationException($"Topic {topic} is invalid!");
192199
}
193200

194-
this.ProcessPayload(topic, payload, SparkplugMetricStatus.Offline);
201+
this.ProcessPayload(topic, filteredMetrics, SparkplugMetricStatus.Offline);
195202
await this.FireDeviceDeathReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifier, topic.DeviceIdentifier);
196203
break;
197204
}
@@ -201,10 +208,10 @@ await this.FireDeviceBirthReceived(topic.GroupIdentifier, topic.EdgeNodeIdentifi
201208
/// Handles the device message.
202209
/// </summary>
203210
/// <param name="topic">The topic.</param>
204-
/// <param name="payload">The payload.</param>
211+
/// <param name="metrics">The metrics.</param>
205212
/// <param name="metricStatus">The metric status.</param>
206213
/// <exception cref="InvalidCastException">Thrown if the metric cast is invalid.</exception>
207-
private IEnumerable<Metric> ProcessPayload(SparkplugMessageTopic topic, Payload payload, SparkplugMetricStatus metricStatus)
214+
private IEnumerable<Metric> ProcessPayload(SparkplugMessageTopic topic, List<Metric> metrics, SparkplugMetricStatus metricStatus)
208215
{
209216
var metricState = new MetricState<Metric>
210217
{
@@ -225,7 +232,7 @@ private IEnumerable<Metric> ProcessPayload(SparkplugMessageTopic topic, Payload
225232
this.NodeStates[topic.EdgeNodeIdentifier] = metricState;
226233
}
227234

228-
foreach (var payloadMetric in payload.Metrics)
235+
foreach (var payloadMetric in metrics)
229236
{
230237
if (payloadMetric is not Metric convertedMetric)
231238
{

src/SparkplugNet/VersionB/SparkplugNode.cs

Lines changed: 10 additions & 6 deletions
Original file line numberDiff line numberDiff line change
@@ -114,9 +114,16 @@ protected override async Task OnMessageReceived(SparkplugMessageTopic topic, byt
114114
private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payload payload)
115115
{
116116
// Filter out session number metric.
117+
var sessionNumberMetric = payload.Metrics.FirstOrDefault(m => m.Name != Constants.SessionNumberMetricName);
117118
var metricsWithoutSequenceMetric = payload.Metrics.Where(m => m.Name != Constants.SessionNumberMetricName);
118-
this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType);
119+
var filteredMetrics = this.KnownMetricsStorage.FilterMetrics(metricsWithoutSequenceMetric, topic.MessageType).ToList();
119120

121+
if (sessionNumberMetric is not null)
122+
{
123+
filteredMetrics.Add(sessionNumberMetric);
124+
}
125+
126+
// Handle messages.
120127
switch (topic.MessageType)
121128
{
122129
case SparkplugMessageType.DeviceCommand:
@@ -125,15 +132,12 @@ private async Task HandleMessagesForVersionB(SparkplugMessageTopic topic, Payloa
125132
throw new InvalidOperationException($"Topic {topic} is invalid!");
126133
}
127134

128-
await this.FireDeviceCommandReceived(topic.DeviceIdentifier, payload.Metrics);
135+
await this.FireDeviceCommandReceived(topic.DeviceIdentifier, filteredMetrics);
129136
break;
130137

131138
case SparkplugMessageType.NodeCommand:
132-
await this.FireNodeCommandReceived(payload.Metrics);
139+
await this.FireNodeCommandReceived(filteredMetrics);
133140
break;
134141
}
135142
}
136-
137-
// Todo: Check exception description in the method description,
138-
// bei casts überall exceptions werfen, wenn es nicht klappt!
139143
}

0 commit comments

Comments
 (0)