![[Jenkins]](/kapua/static/eb5811dc/images/svgs/logo.svg){kind=logo}
Package: KuraPayload
KuraPayload
| name | instruction | branch | complexity | line | method | ||||||||||||||||||||
|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|---|
| KuraPayload() |
|
|
|
|
|
||||||||||||||||||||
| addMetric(String, Object) |
|
|
|
|
|
||||||||||||||||||||
| buildFromProtoBuf(KuraPayloadProto.KuraPayload.KuraPosition) |
|
|
|
|
|
||||||||||||||||||||
| buildPositionProtoBuf(DevicePosition) |
|
|
|
|
|
||||||||||||||||||||
| getBody() |
|
|
|
|
|
||||||||||||||||||||
| getMetrics() |
|
|
|
|
|
||||||||||||||||||||
| getPosition() |
|
|
|
|
|
||||||||||||||||||||
| getProtoKuraMetricValue(KuraPayloadProto.KuraPayload.KuraMetric, KuraPayloadProto.KuraPayload.KuraMetric.ValueType) |
|
|
|
|
|
||||||||||||||||||||
| getTimestamp() |
|
|
|
|
|
||||||||||||||||||||
| lambda$readFromByteArray$1(KuraPayloadProto.KuraPayload.KuraMetric) |
|
|
|
|
|
||||||||||||||||||||
| lambda$toByteArray$0(KuraPayloadProto.KuraPayload.Builder, String, Object) |
|
|
|
|
|
||||||||||||||||||||
| readFromByteArray(byte[]) |
|
|
|
|
|
||||||||||||||||||||
| removeMetric(String) |
|
|
|
|
|
||||||||||||||||||||
| setBody(byte[]) |
|
|
|
|
|
||||||||||||||||||||
| setMetrics(Map) |
|
|
|
|
|
||||||||||||||||||||
| setPosition(DevicePosition) |
|
|
|
|
|
||||||||||||||||||||
| setProtoKuraMetricValue(KuraPayloadProto.KuraPayload.KuraMetric.Builder, Object) |
|
|
|
|
|
||||||||||||||||||||
| setTimestamp(Date) |
|
|
|
|
|
||||||||||||||||||||
| static {...} |
|
|
|
|
|
||||||||||||||||||||
| toByteArray() |
|
|
|
|
|
||||||||||||||||||||
Coverage
1: /*******************************************************************************
2: * Copyright (c) 2016, 2022 Eurotech and/or its affiliates and others
3: *
4: * This program and the accompanying materials are made
5: * available under the terms of the Eclipse Public License 2.0
6: * which is available at https://www.eclipse.org/legal/epl-2.0/
7: *
8: * SPDX-License-Identifier: EPL-2.0
9: *
10: * Contributors:
11: * Eurotech - initial API and implementation
12: *******************************************************************************/
13: package org.eclipse.kapua.service.device.call.message.kura;
14:
15: import com.google.common.primitives.Booleans;
16: import com.google.protobuf.ByteString;
17: import com.google.protobuf.InvalidProtocolBufferException;
18: import org.eclipse.kapua.message.internal.MessageErrorCodes;
19: import org.eclipse.kapua.message.internal.MessageException;
20: import org.eclipse.kapua.service.device.call.message.DevicePayload;
21: import org.eclipse.kapua.service.device.call.message.DevicePosition;
22: import org.eclipse.kapua.service.device.call.message.kura.proto.KuraPayloadProto;
23: import org.eclipse.kapua.service.device.call.message.kura.utils.GZIPUtils;
24: import org.slf4j.Logger;
25: import org.slf4j.LoggerFactory;
26:
27: import java.io.IOException;
28: import java.util.Date;
29: import java.util.HashMap;
30: import java.util.Map;
31:
32: /**
33: * {@link DevicePayload} {@link org.eclipse.kapua.service.device.call.kura.Kura} implementation.
34: *
35: * @since 1.0.0
36: */
37: public class KuraPayload implements DevicePayload {
38:
39: private static final Logger LOG = LoggerFactory.getLogger(KuraPayload.class);
40:
41: /**
42: * The timestamp.
43: *
44: * @since 1.0.0
45: */
46: protected Date timestamp;
47:
48: /**
49: * The {@link DevicePosition}.
50: *
51: * @since 1.0.0
52: */
53: protected DevicePosition position;
54:
55: /**
56: * The metrics.
57: *
58: * @since 1.0.0
59: */
60: protected Map<String, Object> metrics;
61:
62: /**
63: * The raw body.
64: *
65: * @since 1.0.0
66: */
67: protected byte[] body;
68:
69: /**
70: * Constructor.
71: *
72: * @since 1.0.0
73: */
74: public KuraPayload() {
75: super();
76: }
77:
78: @Override
79: public Date getTimestamp() {
80: return timestamp;
81: }
82:
83: @Override
84: public void setTimestamp(Date timestamp) {
85: this.timestamp = timestamp;
86: }
87:
88: @Override
89: public DevicePosition getPosition() {
90: return position;
91: }
92:
93: @Override
94: public void setPosition(DevicePosition position) {
95: this.position = position;
96: }
97:
98: @Override
99: public Map<String, Object> getMetrics() {
100:• if (metrics == null) {
101: metrics = new HashMap<>();
102: }
103:
104: return metrics;
105: }
106:
107: @Override
108: public void setMetrics(Map<String, Object> metrics) {
109: this.metrics = metrics;
110: }
111:
112: @Override
113: public void addMetric(String name, Object value) {
114: getMetrics().put(name, value);
115: }
116:
117: @Override
118: public void removeMetric(String name) {
119: getMetrics().remove(name);
120: }
121:
122: @Override
123: public byte[] getBody() {
124: return body;
125: }
126:
127: @Override
128: public void setBody(byte[] body) {
129: this.body = body;
130: }
131:
132: @Override
133: public byte[] toByteArray() {
134: KuraPayloadProto.KuraPayload.Builder protoMsg = KuraPayloadProto.KuraPayload.newBuilder();
135:
136: //
137: // Set the timestamp
138:• if (getTimestamp() != null) {
139: protoMsg.setTimestamp(getTimestamp().getTime());
140: }
141:
142: //
143: // Set the position
144:• if (getPosition() != null) {
145: protoMsg.setPosition(buildPositionProtoBuf(getPosition()));
146: }
147:
148: //
149: // Set the metrics
150: getMetrics().forEach((name, value) -> {
151:• if (value != null) {
152: try {
153: KuraPayloadProto.KuraPayload.KuraMetric.Builder metricBuilder = KuraPayloadProto.KuraPayload.KuraMetric.newBuilder();
154: metricBuilder.setName(name);
155:
156: setProtoKuraMetricValue(metricBuilder, value);
157: metricBuilder.build();
158:
159: protoMsg.addMetric(metricBuilder);
160: } catch (MessageException me) {
161: LOG.warn("During serialization, ignoring metric named: {}. Unrecognized value type: {}.", name, value.getClass().getName(), me);
162: }
163: }
164: });
165:
166: //
167: // Set the body
168:• if (hasBody()) {
169: protoMsg.setBody(ByteString.copyFrom(getBody()));
170: }
171:
172: return protoMsg.build().toByteArray();
173: }
174:
175: @Override
176: public void readFromByteArray(byte[] bytes) throws MessageException {
177: //
178: // Decompress
179:• if (GZIPUtils.isCompressed(bytes)) {
180: try {
181: bytes = GZIPUtils.decompress(bytes);
182: } catch (IOException ioe) {
183: throw new MessageException(MessageErrorCodes.INVALID_MESSAGE, ioe, (Object[]) null);
184: }
185: }
186:
187: //
188: // Convert protobuf
189: KuraPayloadProto.KuraPayload protoMsg = null;
190: try {
191: protoMsg = KuraPayloadProto.KuraPayload.parseFrom(bytes);
192: } catch (InvalidProtocolBufferException ipbe) {
193: throw new MessageException(MessageErrorCodes.INVALID_MESSAGE, ipbe, (Object[]) null);
194: }
195:
196: //
197: // Add timestamp
198:• if (protoMsg.hasTimestamp()) {
199: timestamp = (new Date(protoMsg.getTimestamp()));
200: }
201:
202: //
203: // Add position
204:• if (protoMsg.hasPosition()) {
205: position = (buildFromProtoBuf(protoMsg.getPosition()));
206: }
207:
208: //
209: // Add metrics
210: protoMsg.getMetricList().forEach(kuraMetric -> {
211: try {
212: Object value = getProtoKuraMetricValue(kuraMetric, kuraMetric.getType());
213: addMetric(kuraMetric.getName(), value);
214: } catch (MessageException me) {
215: LOG.warn("During deserialization, ignoring metric named: {}. Unrecognized value type: {}", kuraMetric.getName(), kuraMetric.getType(), me);
216: }
217: });
218:
219: //
220: // Set the body
221:• if (protoMsg.hasBody()) {
222: setBody(protoMsg.getBody().toByteArray());
223: }
224: }
225:
226: //
227: // Private methods
228: //
229: private Object getProtoKuraMetricValue(KuraPayloadProto.KuraPayload.KuraMetric metric, KuraPayloadProto.KuraPayload.KuraMetric.ValueType type) throws MessageException {
230:• switch (type) {
231: case DOUBLE:
232: return metric.getDoubleValue();
233: case FLOAT:
234: return metric.getFloatValue();
235: case INT64:
236: return metric.getLongValue();
237: case INT32:
238: return metric.getIntValue();
239: case BOOL:
240: return metric.getBoolValue();
241: case STRING:
242: return metric.getStringValue();
243: case BYTES:
244: ByteString bs = metric.getBytesValue();
245: return bs.toByteArray();
246: default:
247: throw new MessageException(MessageErrorCodes.INVALID_METRIC_TYPE, null, type);
248: }
249: }
250:
251: private static void setProtoKuraMetricValue(KuraPayloadProto.KuraPayload.KuraMetric.Builder metric, Object o) throws MessageException {
252:
253:• if (o instanceof String) {
254: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.STRING);
255: metric.setStringValue((String) o);
256:• } else if (o instanceof Double) {
257: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.DOUBLE);
258: metric.setDoubleValue((Double) o);
259:• } else if (o instanceof Integer) {
260: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.INT32);
261: metric.setIntValue((Integer) o);
262:• } else if (o instanceof Float) {
263: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.FLOAT);
264: metric.setFloatValue((Float) o);
265:• } else if (o instanceof Long) {
266: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.INT64);
267: metric.setLongValue((Long) o);
268:• } else if (o instanceof Boolean) {
269: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.BOOL);
270: metric.setBoolValue((Boolean) o);
271:• } else if (o instanceof byte[]) {
272: metric.setType(KuraPayloadProto.KuraPayload.KuraMetric.ValueType.BYTES);
273: metric.setBytesValue(ByteString.copyFrom((byte[]) o));
274:• } else if (o == null) {
275: throw new MessageException(MessageErrorCodes.INVALID_METRIC_VALUE, null, "null");
276: } else {
277: throw new MessageException(MessageErrorCodes.INVALID_METRIC_TYPE, null, o.getClass().getName());
278: }
279: }
280:
281: private KuraPayloadProto.KuraPayload.KuraPosition buildPositionProtoBuf(DevicePosition position) {
282: KuraPayloadProto.KuraPayload.KuraPosition.Builder protoPos = null;
283: protoPos = KuraPayloadProto.KuraPayload.KuraPosition.newBuilder();
284:
285:• if (position.getLatitude() != null) {
286: protoPos.setLatitude(position.getLatitude());
287: }
288:• if (position.getLongitude() != null) {
289: protoPos.setLongitude(position.getLongitude());
290: }
291:• if (position.getAltitude() != null) {
292: protoPos.setAltitude(position.getAltitude());
293: }
294:• if (position.getPrecision() != null) {
295: protoPos.setPrecision(position.getPrecision());
296: }
297:• if (position.getHeading() != null) {
298: protoPos.setHeading(position.getHeading());
299: }
300:• if (position.getSpeed() != null) {
301: protoPos.setSpeed(position.getSpeed());
302: }
303:• if (position.getTimestamp() != null) {
304: protoPos.setTimestamp(position.getTimestamp().getTime());
305: }
306:• if (position.getSatellites() != null) {
307: protoPos.setSatellites(position.getSatellites());
308: }
309:• if (position.getStatus() != null) {
310: protoPos.setStatus(position.getStatus());
311: }
312:
313: return protoPos.build();
314: }
315:
316: private DevicePosition buildFromProtoBuf(KuraPayloadProto.KuraPayload.KuraPosition protoPosition) {
317: DevicePosition devicePosition = getPosition();
318:
319: // For performance reason check the position before
320:• if (devicePosition == null &&
321:• Booleans.countTrue(
322: protoPosition.hasLatitude(),
323: protoPosition.hasLatitude(),
324: protoPosition.hasLongitude(),
325: protoPosition.hasAltitude(),
326: protoPosition.hasPrecision(),
327: protoPosition.hasHeading(),
328: protoPosition.hasHeading(),
329: protoPosition.hasSpeed(),
330: protoPosition.hasSatellites(),
331: protoPosition.hasStatus(),
332: protoPosition.hasTimestamp()) > 0) {
333: devicePosition = new KuraPosition();
334: }
335:
336:• if (devicePosition != null) {
337:• if (protoPosition.hasLatitude()) {
338: devicePosition.setLatitude(protoPosition.getLatitude());
339: }
340:• if (protoPosition.hasLongitude()) {
341: devicePosition.setLongitude(protoPosition.getLongitude());
342: }
343:• if (protoPosition.hasAltitude()) {
344: devicePosition.setAltitude(protoPosition.getAltitude());
345: }
346:• if (protoPosition.hasPrecision()) {
347: devicePosition.setPrecision(protoPosition.getPrecision());
348: }
349:• if (protoPosition.hasHeading()) {
350: devicePosition.setHeading(protoPosition.getHeading());
351: }
352:• if (protoPosition.hasSpeed()) {
353: devicePosition.setSpeed(protoPosition.getSpeed());
354: }
355:• if (protoPosition.hasSatellites()) {
356: devicePosition.setSatellites(protoPosition.getSatellites());
357: }
358:• if (protoPosition.hasStatus()) {
359: devicePosition.setStatus(protoPosition.getStatus());
360: }
361:• if (protoPosition.hasTimestamp()) {
362: devicePosition.setTimestamp(new Date(protoPosition.getTimestamp()));
363: }
364: }
365: return devicePosition;
366: }
367: }