merge tsmatch api into tsmatch engine

TSMatch_Engine/.env

@@ -7,6 +7,19 @@ BROKER_URL=mqtt://10.0.33.39:1883
 BROKER_USERNAME=iiotlabadmin
 BROKER_PASSWORD=iiotlabadmin
 
+BROKER_URL_EXTERNAL=mqtt://10.0.33.39:1883
+BROKER_USERNAME_EXTERNAL=iiotlabadmin
+BROKER_PASSWORD_EXTERNAL=iiotlabadmin
+
+TOPIC_DISCOVERY=INNOVINT/DISCOVERY/NDATA/TSMATCH_INNOVINT_1
+TOPIC_DEADVERTIZE=INNOVINT/DEADVERTIZE/NDATA/TSMATCH_INNOVINT_1
+TOPIC_OBSERVATION=INNOVINT/OBSERVATION/NDATA/TSMATCH_INNOVINT_1
+TOPIC_REQUEST=INNOVINT/HAMBURG_FACTORY1/REQUEST
+TOPIC_DELETE_REQUEST=INNOVINT/HAMBURG_FACTORY1/DELETE_REQUEST
+TOPIC_RESPONSE=INNOVINT/RESPONSE/NDATA/TSMATCH_INNOVINT_1
+TOPIC_REQUEST_DISCOVERY_ALL=INNOVINT/HAMBURG_FACTORY1/REQUEST_DISCOVERY_ALL
+TOPIC_RESPONSE_DISCOVERY_ALL=INNOVINT/DISCOVERY_ALL/NDATA/TSMATCH_INNOVINT_1
+
 NEO4J_HOST=10.0.33.39
 NEO4J_PORT=7687
 NEO4J_DATABASE=neo4j

TSMatch_Engine/src/service/APISensorController.ts

+/**
+ * Copyright (C) 2020 fortiss GmbH
+ * @author Nisrine Bnouhanna – {@link "bnouhanna@fortiss.org"}
+ * @version 1.1
+ * an API that receives service request and publishes observation based on the specified request
+ * Uses coatyio Siemens AG. Licensed under the MIT License
+ */
+import {DbContext} from "@coaty/core/db";
+import {ThingSensorObservationObserverController} from "@coaty/core/sensor-things";
+import {Subject} from "rxjs";
+import {ThingRepository} from "../shared/repository/ThingRepository";
+import {LocationRepository} from "../shared/repository/LocationRepository";
+import {JSONUtil} from "../shared/util/JSONUtil";
+
+export class APISensorController extends ThingSensorObservationObserverController {
+    private _sensorsSubscription = new Subject();
+    private _sensorsDeadvertise = new Subject();
+    private _allSensorsSubscription = new Subject();
+    private _dbCtx: DbContext;
+    private sensorQueue = [];
+    private isWorking = false;
+
+    onInit() {
+        super.onInit();
+        this._dbCtx = new DbContext(this.runtime.databaseOptions["neo4j"]);
+    }
+
+    onCommunicationManagerStarting() {
+        super.onCommunicationManagerStarting();
+
+        // Monitor information about changes in the currently registered sensors.
+        this.registeredSensorsChangeInfo$.subscribe(changeInfo => {
+            if (changeInfo.added.length !== 0) {
+                changeInfo.added.forEach(sensor => {
+                    this.sensorQueue.push(sensor);
+                    if (!this.isWorking) {
+                        this.isWorking = true;
+                        this.observeSensors();
+                    }
+                });
+            } else if (changeInfo.removed.length !== 0) {
+                changeInfo.removed.forEach(sensor => this._sensorsDeadvertise.next(sensor["thing"]));
+            }
+        });
+    }
+
+    GetAllSensorsRaw(rawAllSensors) {
+        rawAllSensors.subscribe(([topic, payload]) => {
+            this._GetAllSensors(payload.toString());
+        });
+    }
+
+    get sensorDeadvertisedObservable() {
+        return this._sensorsDeadvertise.asObservable();
+    }
+
+    get sensorDiscoveryObservable() {
+        return this._sensorsSubscription.asObservable();
+    }
+
+    get allSensorDiscoveryObservable() {
+        return this._allSensorsSubscription.asObservable();
+    }
+
+    /**
+     * Observe newly discovered sensors, get the location/placement of the sensors and
+     * merge it with the sensor description and finally advertise the result
+     */
+    observeSensors() {
+        let sensor = this.sensorQueue.pop();
+        if (!sensor) {
+            this.isWorking = false;
+            return;
+        }
+
+        sensor = JSONUtil.linearize(sensor);
+        // find location of this sensor
+        LocationRepository.getFeatureOfInterest(this._dbCtx, sensor["thing_objectId"]).then(result => {
+            if (result.length > 0) {
+                const location = {
+                    room: result[0]._fields[0],
+                    floor: result[0]._fields[1],
+                    building: {
+                        name: result[0]._fields[2],
+                        address: result[0]._fields[3],
+                    },
+                }
+
+                // advertise the sensor and it's location
+                const payload = {sensor, location};
+                this._sensorsSubscription.next(payload);
+                this.observeSensors();
+            } else {
+                this.sensorQueue.push(sensor);
+                setTimeout(() => this.observeSensors(), 50);
+            }
+        });
+    }
+
+    /**
+     * Get all the registered sensors
+     * @param message
+     */
+    _GetAllSensors(message) {
+        ThingRepository.getAllSensors(this._dbCtx).then(result => {
+            result.forEach(sensorData => {
+                let sensor = sensorData._fields[0].properties;
+
+                // find the location info for each of the sensor
+                LocationRepository.getFeatureOfInterest(this._dbCtx, sensor.parentObjectId).then(result => {
+                    const location = {
+                        room: result[0]._fields[0],
+                        floor: result[0]._fields[1],
+                        building: {
+                            name: result[0]._fields[2],
+                            address: result[0]._fields[3],
+                        },
+                    }
+
+                    // publish the sensors along with their location
+                    const payload = {sensor, location};
+                    this._allSensorsSubscription.next(payload);
+                });
+            })
+        });
+        console.log("Request for all sensor available received !!");
+    }
+}

TSMatch_Engine/src/service/APITaskController.ts

+/**
+ * Copyright (C) 2020 fortiss GmbH
+ * @author Nisrine Bnouhanna {@link "bnouhanna@fortiss.org"}
+ * @version 1.1
+ * an API that receives service request and publishes observation based on the specified request
+ * Uses coatyio Siemens AG. Licensed under the MIT License
+ */
+
+import {AdvertiseEvent, Controller, TaskStatus} from "@coaty/core";
+import {DbContext} from "@coaty/core/db";
+import {Subject} from "rxjs";
+import {
+    DatabaseChange,
+    modelTypes,
+    ObservationResults,
+    ServiceTaskRequest,
+    ServieTaskRequestPriority
+} from "../shared/models";
+import {RequestRepository} from "../shared/repository/RequestRepository";
+import {GroupingRepository} from "../shared/repository/GroupingRepository";
+import {LocationRepository} from "../shared/repository/LocationRepository";
+
+
+/**
+ * Generate a service request based on user input then advertize it as object request
+ */
+export class APITaskController extends Controller {
+    serviceRequest;
+    private _dbCtx: DbContext;
+    private _response = new Subject();
+    private _results = new Subject();
+
+    onInit() {
+        super.onInit();
+        this._dbCtx = new DbContext(this.runtime.databaseOptions["neo4j"]);
+    }
+
+    onCommunicationManagerStarting() {
+        super.onCommunicationManagerStarting();
+        this._observeMatchingResults();
+    }
+
+    registerRaw(rawObs) {
+        rawObs.subscribe(([topic, payload]) => {
+            this._generateRequests(payload.toString());
+        });
+    }
+
+    unsubscribeRequestRaw(rawUnsubscribeRaw) {
+        rawUnsubscribeRaw.subscribe(([topic, payload]) => {
+            this._deleteRequest(payload.toString());
+        });
+    }
+
+    get responseObservable() {
+        return this._response.asObservable();
+    }
+
+    get resultsObservable() {
+        return this._results.asObservable();
+    }
+
+    /**
+     * Observe grouping events and return response to the users
+     * @private
+     */
+    private _observeMatchingResults() {
+        this.communicationManager
+            .observeAdvertiseWithObjectType(modelTypes.OBJECT_TYPE_DATABASE_CHANGE)
+            .subscribe(event => {
+                if (event.data.object.name !== "GROUPING_CHANGED") {
+                    return;
+                }
+
+                // send successful grouping results
+                GroupingRepository.getNotSentSuccessfulGroupings(this._dbCtx).then(result => {
+                    let groupingList = {};
+                    let requestIdList = [];
+                    let matchedLocations = {};
+                    result.forEach(grouping => {
+                        // create a response object for each of the grouping
+                        const requestId = grouping._fields[0].properties.objectId;
+                        const sensorId = grouping._fields[1].properties.objectId;
+                        const aggregationId = grouping._fields[2].properties.aggregationId;
+                        if (!groupingList[requestId]) {
+                            groupingList[requestId] = {};
+                            groupingList[requestId][aggregationId] = [];
+                            groupingList[requestId][aggregationId].push(sensorId);
+                            matchedLocations[requestId] = [];
+                        } else {
+                            if (!groupingList[requestId][aggregationId]) {
+                                groupingList[requestId][aggregationId] = [];
+                            }
+                            if (!groupingList[requestId][aggregationId].includes(sensorId)) {
+                                groupingList[requestId][aggregationId].push(sensorId);
+                            }
+                        }
+
+                        let location = {
+                            building: {
+                                name: grouping._fields[5].properties.name,
+                                address: grouping._fields[5].properties.address
+                            },
+                            floor: grouping._fields[4].properties.name,
+                            room: grouping._fields[3].properties.name
+                        }
+
+                        // find the registered locations with a related sensor for the given service request
+                        let exists = false;
+                        matchedLocations[requestId].forEach(locationElement => {
+                            if (JSON.stringify(locationElement) === JSON.stringify(location)) {
+                                exists = true;
+                            }
+                        })
+                        if (!exists) {
+                            matchedLocations[requestId].push(location);
+                        }
+
+                        requestIdList.push(requestId);
+                        GroupingRepository.markGroupingAsResponseSent(this._dbCtx, requestId).then();
+                    });
+
+                    // find the locations that has no sensor related with the given service request
+                    this._findNonMatchedLocations(matchedLocations).then(nonMatchedLocationList => {
+                        this._response.next({
+                            grouping: groupingList,
+                            nonMatchedLocations: nonMatchedLocationList
+                        });
+                    });
+
+                    // get unsuccessful groupings that are not yet sent as response (requests)
+                    GroupingRepository.getNotSentUnsuccessfulGroupings(this._dbCtx).then(result => {
+                        result.forEach(grouping => {
+                            let response = {};
+                            const requestId = grouping._fields[0].properties.objectId;
+                            if (!requestIdList.includes(requestId)) {
+                                response["name"] = "groupingNotPossible";
+                                response["requestId"] = requestId;
+
+                                GroupingRepository.markGroupingAsResponseSent(this._dbCtx, requestId);
+                                this._response.next(response);
+                            }
+                        });
+                    })
+                });
+            });
+
+        this.communicationManager
+            // Observe advertised event of type Database Change sent when a change in the DB collection grouping occurs
+            .observeAdvertiseWithObjectType(modelTypes.OBJECT_TYPE_OBSERVATION_RESULTS)
+            .subscribe(event => {
+                const j = event.data.object as ObservationResults;
+                let sensorIdList = [];
+                j.grouping.forEach(group => {
+                    sensorIdList.push(group.sensorId);
+                })
+                const obs = {
+                    name: j.name,
+                    objectId: j.objectId,
+                    objectType: j.objectType,
+                    responseId: j.grouping[0].requestObjectId,
+                    results: j.results,
+                    timestamp: j.timestamp,
+                    sensorIds: sensorIdList,
+                    location: j.grouping[0].location
+                };
+                this._results.next(obs);
+            });
+    }
+
+    /**
+     * find nonMatched location for each of the service request given in the "matchedLocations" parameter
+     * matched location refers to a location node in the db with a sensor that matches the requirements
+     * of that request and therefore provide data
+     * @param matchedLocations
+     * @private
+     */
+    private async _findNonMatchedLocations(matchedLocations) {
+        let nonMatchedLocations = {};
+
+        await LocationRepository.getAllLocationNodes(this._dbCtx).then(locationList => {
+            locationList.forEach(locationNode => {
+                // get each of the location nodes in the db
+                let locationObject = {
+                    building: {
+                        name: locationNode._fields[2].properties.name,
+                        address: locationNode._fields[2].properties.address
+                    },
+                    floor: locationNode._fields[1].properties.name,
+                    room: locationNode._fields[0].properties.name,
+                }
+
+                // for each of the service request in matchedLocations, find the non-matched locations
+                for (let requestId in matchedLocations) {
+                    let exists = false;
+                    if (!nonMatchedLocations[requestId]) {
+                        nonMatchedLocations[requestId] = [];
+                    }
+                    matchedLocations[requestId].forEach(matchedLocation => {
+                        if (JSON.stringify(matchedLocation) === JSON.stringify(locationObject)) {
+                            exists = true;
+                        }
+                    });
+
+                    if (!exists) {
+                        nonMatchedLocations[requestId].push(locationObject);
+                    }
+                }
+
+            });
+        });
+
+        return nonMatchedLocations;
+    }
+
+    /**
+     * Generate a service request and advertise it over coaty
+     * @param message
+     * @private
+     */
+    private _generateRequests(message) {
+        message = JSON.parse(message);
+        if (message.hasOwnProperty("featureOfInterest") && message.hasOwnProperty("unitOfMeasurement") && message.hasOwnProperty("creatorId") && message.hasOwnProperty("threshold") && message.hasOwnProperty("location") && message.hasOwnProperty("uuid") && message.uuid !== undefined && message.location !== undefined && message.threshold !== undefined && message.creatorId !== undefined && message.unitOfMeasurement !== undefined && message.featureOfInterest !== undefined) {
+            // tslint:disable-next-line: max-line-length
+            let serviceRequest = this._createRequest(message);
+            RequestRepository.insertRequest(this._dbCtx, serviceRequest).then(result => {
+                this.communicationManager.publishAdvertise(AdvertiseEvent.withObject(serviceRequest));
+            });
+            console.log("|-------------------------------------------------------------------------------------|");
+            console.log("|-----> Your service request has been advertised: ------------------------------------|");
+            console.log(serviceRequest);
+        } else {
+            console.log("service request is not valid");
+        }
+    }
+
+    /**
+     * Creates a new service request
+     */
+    private _createRequest(rawRequest): ServiceTaskRequest {
+        return {
+            objectId: rawRequest.uuid,
+            objectType: modelTypes.OBJECT_TYPE_SERVICE_DESCRIPTION,
+            coreType: "Task",
+            name: `Service Request`,
+            creatorId: rawRequest.creatorId,
+            creationTimestamp: Date.now(),
+            status: TaskStatus.Request,
+            FeatureOfInterest: rawRequest.featureOfInterest,
+            unitOfMeasurement: rawRequest.unitOfMeasurement,
+            threshold: rawRequest.threshold,
+            Location: rawRequest.location,
+            MQTTtopic: "updates",
+            urgency: ServieTaskRequestPriority.Critical,
+            domain: rawRequest.domain,
+            domainDetail: rawRequest.domainDetail,
+            monitoringAspect: rawRequest.monitoringAspect,
+            monitoringAspectDetail: rawRequest.monitoringAspectDetail
+        };
+    }
+
+    /**
+     * Delete a given service request and advertise the deletion
+     * @param message
+     * @private
+     */
+    private _deleteRequest(message) {
+        message = JSON.parse(message);
+        console.log("Delete request: ", message);
+        if (message.hasOwnProperty("requestId") && message.requestId !== undefined) {
+            RequestRepository.deleteRequest(this._dbCtx, message.requestId).then(result => {
+                console.log("Request deleted!");
+                this._advertiseDatabaseChange();
+            }).catch(error => {
+                console.log("Error while deleting a request: ", message.requestId, " - ", error);
+            });
+        } else {
+            console.log("paylaod of DELETE_REQUEST is not valid");
+        }
+    }
+
+    // Publish an DB change of the newly created grouping
+    private _advertiseDatabaseChange() {
+        const change: DatabaseChange = {
+            name: "GROUPING_CHANGED",
+            objectId: this.runtime.newUuid(),
+            objectType: modelTypes.OBJECT_TYPE_DATABASE_CHANGE,
+            coreType: "CoatyObject",
+            hasLogChanged: true,
+            hasTaskChanged: true,
+        };
+        this.communicationManager.publishAdvertise(AdvertiseEvent.withObject(change));
+    }
+}

TSMatch_Engine/src/service/Agent.ts

@@ -15,7 +15,11 @@ import {agentInfo} from "./agent.info";
 import {FeatureOfInterestController} from "./FeatureOfInterestController";
 import {SensorController} from "./SensorController";
 import {TaskController} from "./TaskController";
+import {APITaskController} from "./APITaskController";
+import {APISensorController} from "./APISensorController";
 import {BaseRepository} from "../shared/repository/BaseRepository";
+import {RawController} from "./RawController";
+import {serviceConfigRaw} from "../shared/configRaw";
 
 
 const components: Components = {
@@ -23,9 +27,26 @@ const components: Components = {
         TaskController,
         SensorController,
         FeatureOfInterestController,
+        APITaskController,
+        APISensorController
     },
 };
 
+const componentsRaw: Components = {
+    controllers: {
+        RawController,
+    },
+};
+const configurationRaw: Configuration = mergeConfigurations(
+    serviceConfigRaw(),
+    {
+        controllers: {
+            RawController: {
+                requestGenerationInterval: 10000,
+            },
+        },
+    });
+
 const configuration: Configuration = mergeConfigurations(
     serviceConfig(agentInfo, "Service"),
     {
@@ -63,6 +84,12 @@ const configuration: Configuration = mergeConfigurations(
                 shouldPersistObservedSnapshots: false,
                 shouldReplyToQueries: true,
             },
+            APITaskController: {
+                requestGenerationInterval: 10000,
+            },
+            APISensorController: {
+                requestGenerationInterval: 10000,
+            },
         },
     });
 
@@ -83,5 +110,21 @@ BaseRepository.initializeDatabase(configuration.databases)
 
         // Log broker connection state changes (online/offline) to the console.
         NodeUtils.logCommunicationState(container);
+        const containerRaw = Container.resolve(componentsRaw, configurationRaw);
+        NodeUtils.logCommunicationState(containerRaw);
+
+        const rawController = containerRaw.getController<RawController>("RawController");
+        const taskController = container.getController<APITaskController>("APITaskController");
+        const sensorController = container.getController<APISensorController>("APISensorController");
+
+        // register tsmatch api tasks
+        taskController.registerRaw(rawController.rawObservable);
+        taskController.unsubscribeRequestRaw(rawController.deleteRequestObservable);
+        sensorController.GetAllSensorsRaw(rawController.AllDiscoveredSensors);
+        rawController.registerResponse(taskController.responseObservable);
+        rawController.registerResults(taskController.resultsObservable);
+        rawController.registerSensorDiscovery(sensorController.sensorDiscoveryObservable);
+        rawController.registerSensorDeadvertize(sensorController.sensorDeadvertisedObservable);
+        rawController.registerAllSensorDiscovery(sensorController.allSensorDiscoveryObservable);
     })
     .catch(error => NodeUtils.logError(error, "Failed to initialize database."));

TSMatch_Engine/src/service/RawController.ts

+/**
+ * Copyright (C) 2020 fortiss GmbH
+ * @author Nisrine Bnouhanna – {@link "bnouhanna@fortiss.org"}
+ * @version 1.1
+ * an API that receives service request and publishes observation based on the specified request
+ * Uses coatyio Siemens AG. Licensed under the MIT License
+ */
+
+import {Controller} from "@coaty/core";
+import {NodeUtils} from "@coaty/core/runtime-node";
+import {Subject} from "rxjs";
+
+
+/**
+ * Generate a service request based on user input then advertize it as object request
+ */
+export class RawController extends Controller {
+
+    private _rawSubject = new Subject();
+    private _deleteRequest = new Subject();
+    private _AllDiscoveredSensors = new Subject();
+
+    onInit() {
+        super.onInit();
+        if (!process.env.TOPIC_DISCOVERY && !process.env.TOPIC_OBSERVATION && !process.env.TOPIC_REQUEST && !process.env.TOPIC_RESPONSE) {
+            NodeUtils.logError(new Error("Missing MQTT topics"), "MQTT topics not specified.");
+            process.exit(1);
+        }
+    }
+
+    /**
+     * Subscribe to service request, service request deletion and discovering all sensors topics
+     */
+    onCommunicationManagerStarting() {
+        super.onCommunicationManagerStarting();
+        this.communicationManager
+            .observeRaw(process.env.TOPIC_REQUEST)
+            .subscribe(topicWithPayload => this._rawSubject.next(topicWithPayload));
+        this.communicationManager
+            .observeRaw(process.env.TOPIC_DELETE_REQUEST)
+            .subscribe(topicWithPayload => this._deleteRequest.next(topicWithPayload));
+        this.communicationManager
+            .observeRaw(process.env.TOPIC_REQUEST_DISCOVERY_ALL)
+            .subscribe(topicWithPayload => this._AllDiscoveredSensors.next(topicWithPayload));
+    }
+
+    get AllDiscoveredSensors() {
+        return this._AllDiscoveredSensors.asObservable();
+    }
+
+    get deleteRequestObservable() {
+        return this._deleteRequest.asObservable();
+    }
+
+    get rawObservable() {
+        return this._rawSubject.asObservable();
+    }
+
+    /**
+     * Subscribe to the raw response topic and advertise the messages
+     * @param rawResponse
+     */
+    registerResponse(rawResponse) {
+        rawResponse.subscribe((response) => {
+            this.communicationManager
+                .publishRaw(process.env.TOPIC_RESPONSE, JSON.stringify(response));
+        });
+    }
+
+    /**
+     * Subscribe to the sensor observation topic and advertise the messages
+     * @param rawResults
+     */
+    registerResults(rawResults) {
+        rawResults.subscribe((results) => {
+            this.communicationManager
+                .publishRaw(process.env.TOPIC_OBSERVATION, JSON.stringify(results));
+        });
+    }
+
+    /**
+     * Subscribe to the sensor discovery topic and advertise the messages
+     * @param rawSensorDiscovery
+     */
+    registerSensorDiscovery(rawSensorDiscovery) {
+        rawSensorDiscovery.subscribe((sensorDiscovery) => {
+            console.log("Sensor discovered");
+            this.communicationManager
+                .publishRaw(process.env.TOPIC_DISCOVERY, JSON.stringify(sensorDiscovery));
+        });
+    }
+
+    /**
+     * Subscribe to the sensor deadvertisement topic and advertise the messages
+     * @param rawSensorDeadvertise
+     */
+    registerSensorDeadvertize(rawSensorDeadvertise) {
+        rawSensorDeadvertise.subscribe((sensorDeadvertize) => {
+            console.log("Sensor deadvertised");
+            this.communicationManager
+                .publishRaw(process.env.TOPIC_DEADVERTIZE, JSON.stringify(sensorDeadvertize));
+        });
+    }
+
+    /**
+     * Subscribe to the topic which all the sensors are published at once and advertise the messages
+     * @param rawAllSensorDiscovery
+     */
+    registerAllSensorDiscovery(rawAllSensorDiscovery) {
+        rawAllSensorDiscovery.subscribe((allSensorDiscovery) => {
+            console.log("All available sensor published ");
+            this.communicationManager
+                .publishRaw(process.env.TOPIC_RESPONSE_DISCOVERY_ALL, JSON.stringify(allSensorDiscovery));
+        });
+    }
+}

TSMatch_Engine/src/shared/configRaw.ts

+/**
+ * Copyright (C) 2020 fortiss GmbH
+ * @author Nisrine Bnouhanna – {@link "bnouhanna@fortiss.org"}
+ * @version 1.1
+ * the config file for the coaty conatiner specifies the broker URL and initializes the DB
+ * Uses coatyio Siemens AG. Licensed under the MIT License
+ */
+
+import { Configuration} from "@coaty/core";
+import { NodeUtils } from "@coaty/core/runtime-node";
+// import * as fs from "fs";
+// const brokerURL = "mqtts://dataspine.efpf.linksmart.eu:8883";
+// const brokerURL = "mqtt://broker.hivemq.com:1883";
+// import * as path from "path";
+NodeUtils.logInfo(`BROKER_URL=${process.env.BROKER_URL_EXTERNAL}`);
+// const CERT = fs.readFileSync(path.resolve("./src/shared/cacert-2020-07-22.pem"));
+
+if (!process.env.BROKER_URL_EXTERNAL) {
+    NodeUtils.logError(new Error("Missing Broker URL"), "Environment variable BROKER_URL not specified.");
+    process.exit(1);
+}
+
+/**
+ * Gets common Configuration object for TSMatch App
+ *
+ * @param agentInfo the component's agent info
+ */
+export function serviceConfigRaw(): Configuration {
+    return {
+        communication: {
+            brokerUrl: process.env.BROKER_URL_EXTERNAL,
+            mqttClientOptions : {
+                username: process.env.BROKER_USERNAME_EXTERNAL,
+                password: process.env.BROKER_PASSWORD_EXTERNAL,
+                // username: "for:bnouhanna@fortiss.org",
+                // password: "ARFiKh71xR5nTSQSPMnSS9BB",
+                // rejectUnauthorized: false,
+                // cert: CERT,
+                // protocol: "mqtts",
+            },
+            shouldAutoStart: true,
+        },
+    };
+}

TSMatch_Engine/src/shared/repository/GroupingRepository.ts

@@ -17,8 +17,7 @@ export class GroupingRepository {
      */
     public static getNonProcessedGroupings(dbContext: DbContext) {
         return dbContext.callExtension("runQuery",
-            `match (r:Request {isProcessed: false})<-[rel:USED_BY]-(s:Sensor)<-[:HAD_MEMBER]-(t:Thing)-[:AT_LOCATION]->
-            (r2:Room)-[:AT_LOCATION]->(f:Floor)-[:AT_LOCATION]->(b:Building) return r,s,rel,r2.name,f.name,b.name,b.address`);
+            `match (r:Request {isProcessed: false})<-[rel:USED_BY]-(s:Sensor) return r,s,rel`);
     }
 
     /**
@@ -32,17 +31,24 @@ export class GroupingRepository {
     }
 
     /**
-     * Mark a given request as processed by the tsmatch engine
+     * Get all the groupings that are processed by the tsmatch engine but a response isn't sent to the user for that grouping
      * @param dbContext
-     * @param requestObjectId
-     * @param sensorObjectId
      */
-    public static markRequestAsProcessed(dbContext: DbContext, requestObjectId: string, sensorObjectId: string) {
+    public static getNotSentSuccessfulGroupings(dbContext: DbContext) {
         return dbContext.callExtension("runQuery",
-            `match (r:Request {objectId: $requestObjectId, isProcessed: true})<-[rel:USED_BY]-(s:Sensor {objectId: $sensorObjectId}) return r,s,rel`, {
-                requestObjectId: requestObjectId,
-                sensorObjectId: sensorObjectId
-            });
+            `match (r:Request {responseSent: false, isProcessed: true})<-[rel:USED_BY]-(s:Sensor)<-[:HAD_MEMBER]-(t:Thing)-[:AT_LOCATION]
+            ->(room:Room)-[:AT_LOCATION]->(f:Floor)-[:AT_LOCATION]->(b:Building) return r,s,rel,room,f,b`);
+    }
+
+    /**
+     * Get all the unsuccessful groupings that are not yet sent to the user
+     * @param dbContext
+     */
+    public static getNotSentUnsuccessfulGroupings(dbContext: DbContext) {
+        return dbContext.callExtension("runQuery",
+            `match (r:Request {responseSent: false, isProcessed: true})
+            where not (r)<-[:USED_BY]-(:Sensor)
+            return r`);
     }
 
     /**
@@ -50,12 +56,11 @@ export class GroupingRepository {
      * @param dbContext
      * @param requestObjectId
      */
-    public static markGroupingAsResponseSent(dbContext: DbContext, requestObjectId: string, sensorObjectId: string) {
+    public static markGroupingAsResponseSent(dbContext: DbContext, requestObjectId: string) {
         return dbContext.callExtension("runQuery",
-            `match (r:Request {objectId: $requestObjectId, responseSent: false})<-[rel:USED_BY]-(s:Sensor {objectId: $sensorObjectId})
+            `match (r:Request {objectId: $requestObjectId})
             set r.responseSent=true`, {
-                requestObjectId: requestObjectId,
-                sensorObjectId: sensorObjectId
+                requestObjectId: requestObjectId
             });
     }
 }

TSMatch_Engine/src/shared/repository/LocationRepository.ts

@@ -55,17 +55,28 @@ export class LocationRepository {
             return buildingAddress, floorName, roomName, thing, sensor `);
     }
 
+
+
     /**
      * Get featureOfInterest(location) of a given iot thing
      * @param dbContext
-     * @param thingId
+     * @param thingObjectId
      */
-    static getFeatureOfInterest(dbContext: DbContext, thingId: string) {
+    static getFeatureOfInterest(dbContext: DbContext, thingObjectId: string) {
         return dbContext.callExtension("runQuery",
-            `match (t:Thing {objectId: $thingId})
+            `match (t:Thing {objectId: $thingObjectId})
             -[:AT_LOCATION]->(r:Room)-[:AT_LOCATION]->(f:Floor)-[:AT_LOCATION]->(b:Building)
-            return t.objectId, r.name, f.name, b.name`, {
-                thingId: thingId
+            return r.name, f.name, b.name, b.address`, {
+                thingObjectId: thingObjectId
             });
     }
+
+    /**
+     * get a list of all location nodes
+     * @param dbContext
+     */
+    static getAllLocationNodes(dbContext: DbContext) {
+        return dbContext.callExtension("runQuery",
+            `match (r:Room)-[:AT_LOCATION]->(f:Floor)-[:AT_LOCATION]->(b:Building) return r,f,b`);
+    }
 }

TSMatch_Engine/src/shared/repository/RequestRepository.ts

+/**
+ * Copyright (C) 2021 fortiss GmbH
+ * @author Erkan Karabulut – {@link "karabulut@fortiss.org"}
+ * @version 1.1
+ * DB operations related service requests
+ */
+
+import {DbContext} from "@coaty/core/db";
+
+/**
+ * Service request related service requests
+ */
+export class RequestRepository {
+
+    /**
+     * Insert a given request to the db
+     * @param dbContext
+     * @param request
+     */
+    static insertRequest(dbContext: DbContext, request: {}) {
+        // mark the request as not processed yet
+        request["responseSent"] = false;
+        request["isProcessed"] = false;
+        return dbContext.callExtension("runQuery",
+            `with apoc.convert.fromJsonMap($request) as requestObject
+            create (r:Request)
+            set r=requestObject`, {
+                request: JSON.stringify(request)
+            }
+        );
+    }
+
+    /**
+     * Delete a request by ID and also the relations(edges) of it
+     * @param dbContext
+     * @param requestId
+     */
+    static deleteRequest(dbContext: DbContext, requestId) {
+        return dbContext.callExtension("runQuery",
+            `match (t)-[rel]-(r:Request {objectId: $objectId})
+            delete rel,r`, {
+                objectId: requestId
+            });
+    }
+
+    /**
+     * get a list of requests
+     * @param dbContext
+     */
+    static getAllRequests(dbContext: DbContext) {
+        return dbContext.callExtension("runQuery",
+            `match (r:Request) return (r)`);
+    }
+}