Merge branch 'stage' into master for release

Author: fcv-iteratorIt

src/modules/open-data-dk-sharing.module.ts

@@ -4,9 +4,10 @@ import { OpenDataDkSharingService } from "@services/data-management/open-data-dk
 import { SharedModule } from "@modules/shared.module";
 import { OrganizationModule } from "@modules/user-management/organization.module";
 import { PayloadDecoderExecutorModuleModule } from "@modules/payload-decoder-executor-module.module";
+import { ChirpstackAdministrationModule } from "@modules/device-integrations/chirpstack-administration.module";
 
 @Module({
-    imports: [SharedModule, OrganizationModule, PayloadDecoderExecutorModuleModule],
+    imports: [SharedModule, OrganizationModule, PayloadDecoderExecutorModuleModule, ChirpstackAdministrationModule],
     controllers: [OpenDataDkSharingController],
     providers: [OpenDataDkSharingService],
 })

src/services/data-management/open-data-dk-sharing.service.ts

@@ -12,13 +12,16 @@ import { IoTDevice } from "@entities/iot-device.entity";
 import configuration from "@config/configuration";
 import { OpenDataDkSharingController } from "@admin-controller/open-data-dk-sharing.controller";
 import { ErrorCodes } from "@enum/error-codes.enum";
+import { IoTDeviceType } from "@enum/device-type.enum";
+import { ChirpstackDeviceService } from "@services/chirpstack/chirpstack-device.service";
 
 @Injectable()
 export class OpenDataDkSharingService {
     constructor(
         @InjectRepository(OpenDataDkDataset)
         private repository: Repository<OpenDataDkDataset>,
-        private payloadDecoderExecutorService: PayloadDecoderExecutorService
+        private payloadDecoderExecutorService: PayloadDecoderExecutorService,
+        private chirpstackDeviceService: ChirpstackDeviceService
     ) {}
 
     private readonly BACKEND_BASE_URL = configuration()["backend"]["baseurl"];
@@ -40,18 +43,17 @@ export class OpenDataDkSharingService {
             return { error: ErrorCodes.NoData };
         }
 
-        return this.decodeData(rawData);
+        return await this.decodeData(rawData);
     }
 
-    private decodeData(rawData: OpenDataDkDataset) {
-        // TODO: Do this in parallel
+    private async decodeData(rawData: OpenDataDkDataset) {
         const results: any[] = [];
-        rawData.dataTarget.connections.forEach(connection => {
+        for (const connection of rawData.dataTarget.connections) {
             this.logger.debug(`Got connection(${connection.id})`);
-            connection.iotDevices.forEach(async device => {
+            for (const device of connection.iotDevices) {
                 await this.decodeDevice(device, connection, results);
-            });
-        });
+            }
+        }
         return results;
     }
 
@@ -67,6 +69,14 @@ export class OpenDataDkSharingService {
         }
         try {
             if (connection.payloadDecoder != null) {
+                // Enrich lorawan devices with chirpstack data
+                if (
+                    device.type === IoTDeviceType.LoRaWAN &&
+                    connection.payloadDecoder.decodingFunction.includes("lorawanSettings")
+                ) {
+                    device = await this.chirpstackDeviceService.enrichLoRaWANDevice(device);
+                }
+
                 const decoded = await this.payloadDecoderExecutorService.callUntrustedCode(
                     connection.payloadDecoder.decodingFunction,
                     device,

src/services/data-targets/data-target-kafka-listener.service.ts

@@ -44,23 +44,16 @@ export class DataTargetKafkaListenerService extends AbstractKafkaConsumer {
         try {
             iotDevice = await this.ioTDeviceService.findOne(dto.iotDeviceId);
         } catch (err) {
-            this.logger.error(
-                `Error finding IoTDevice by id: ${dto.iotDeviceId}. Stopping.`
-            );
+            this.logger.error(`Error finding IoTDevice by id: ${dto.iotDeviceId}. Stopping.`);
             return;
         }
 
-        this.logger.debug(
-            `Sending payload from deviceId: ${iotDevice.id}; Name: '${iotDevice.name}'`
-        );
+        this.logger.debug(`Sending payload from deviceId: ${iotDevice.id}; Name: '${iotDevice.name}'`);
 
         await this.findDataTargetsAndSend(iotDevice, dto);
     }
 
-    private async findDataTargetsAndSend(
-        iotDevice: IoTDevice,
-        dto: TransformedPayloadDto
-    ) {
+    private async findDataTargetsAndSend(iotDevice: IoTDevice, dto: TransformedPayloadDto) {
         // Get connections in order to only send to the dataTargets which is identified by the pair of IoTDevice and PayloadDecoder
         const dataTargets = await this.dataTargetService.findDataTargetsByConnectionPayloadDecoderAndIoTDevice(
             iotDevice.id,
@@ -79,13 +72,9 @@ export class DataTargetKafkaListenerService extends AbstractKafkaConsumer {
             if (target.type == DataTargetType.HttpPush) {
                 try {
                     const status = await this.httpPushDataTargetService.send(target, dto);
-                    this.logger.debug(
-                        `Sent to HttpPush target: ${JSON.stringify(status)}`
-                    );
+                    this.logger.debug(`Sent to HttpPush target: ${JSON.stringify(status)}`);
                 } catch (err) {
-                    this.logger.error(
-                        `Error while sending to Http Push DataTarget: ${err}`
-                    );
+                    this.logger.error(`Error while sending to Http Push DataTarget: ${err}`);
                 }
             } else if (target.type == DataTargetType.Fiware) {
                 try {
@@ -100,8 +89,10 @@ export class DataTargetKafkaListenerService extends AbstractKafkaConsumer {
                 } catch (err) {
                     this.logger.error(`Error while sending to MQTT DataTarget: ${err}`);
                 }
+            } else if (target.type === DataTargetType.OpenDataDK) {
+                // OpenDataDk data targets are handled uniquely and ignored here.
             } else {
-                throw new NotImplementedException(`Not implemented for: ${target.type}`);
+                this.logger.error(`Not implemented for: ${target.type}`);
             }
         });
     }