From 4c7417d6471b81d8e20f43596ab452471bd813e0 Mon Sep 17 00:00:00 2001 From: Daniel Kluge Date: Mon, 30 Oct 2023 14:25:43 +0100 Subject: [PATCH] Kinda first ready? --- .../example_modules/httpInterfaceServer.ts | 48 +++++--- lib/src/example_modules/mqttConnector.ts | 21 ++-- lib/src/interfaceConnectionObject.ts | 27 +++-- lib/src/multimessageBroker.ts | 108 +++++++++++++++--- lib/src/server.ts | 2 +- lib/src/types/aidConf.ts | 2 +- lib/src/types/requests.ts | 86 +++++++++++++- 7 files changed, 235 insertions(+), 59 deletions(-) diff --git a/lib/src/example_modules/httpInterfaceServer.ts b/lib/src/example_modules/httpInterfaceServer.ts index af90a35..e63f4e2 100644 --- a/lib/src/example_modules/httpInterfaceServer.ts +++ b/lib/src/example_modules/httpInterfaceServer.ts @@ -2,7 +2,7 @@ import type { Express } from "express"; import express from "express"; import { jsonization, types } from "@aas-core-works/aas-core3.0-typescript"; import AASInterfaceServer from "../server"; -import type { OnRequestCallback } from "types/requests"; +import type { OnRequestCallback, Request } from "types/requests"; import Traverser from "../helper/traverser"; export type Config = { @@ -125,7 +125,7 @@ export default class HTTPInterfaceServer extends AASInterfaceServer { idShortPathString.endsWith("$path")) return res.status(501).end(); // If operation ids are changed, you can change the format here - const endingMatchSearch = idShortPathString.match(/(\/attachment$|\/invoke$|\/invoke-async$|\/operation-status\/[a-zA-Z0-9\-]+$|\/operation-result\/[a-zA-Z0-9\-]+$)/) + const endingMatchSearch = idShortPathString.match(/(\/value$|\/attachment$|\/invoke$|\/invoke-async$|\/operation-status\/[a-zA-Z0-9\-]+$|\/operation-result\/[a-zA-Z0-9\-]+$)/) const endingMatch = endingMatchSearch?.[0]; const idShortPath = endingMatch ? idShortPathString.replace(new RegExp(`${endingMatch}$`), "").split("/") : idShortPathString.split("/"); @@ -135,31 +135,46 @@ export default class HTTPInterfaceServer extends AASInterfaceServer { case undefined: { if (req.method !== "GET") return res.status(501).end(); const prop = Traverser.getElementByIdPath(this.aas, sm, idShortPath); + if (prop === null || !types.isProperty(prop)) return res.status(404).end(); - // TODO - //! Check if mapping and live data is available! - - if (prop === null) return res.status(404).end(); return res.json(jsonization.toJsonable(prop)).end(); } + case "/value": + if (req.method !== "GET" && req.method !== "PUT") return res.status(501).end(); + const prop = Traverser.getElementByIdPath(this.aas, sm, idShortPath); + if (prop === null || !types.isProperty(prop)) return res.status(404).end(); + + const request: Request = req.method === "GET" ? { type: "READ", target: prop } : { type: "WRITE", target: prop, extraData: { value: req.body } }; + + const value = this.onRequestCallback(request); + + if (req.method === "PUT") res.status(204).end(); + + if (value === null) return res.status(500).end(); + return res.json(value).end(); case "/attachment": return res.status(501).end(); case "/invoke": { if (req.method !== "POST") return res.status(405).end(); + const op = Traverser.getElementByIdPath(this.aas, sm, idShortPath); + if (op === null || !types.isOperation(op)) return res.status(404).end(); // TODO - // Mapping, dann Befehl // Parameter? + const success = this.onRequestCallback({ type: "CALL", target: op, extraData: { args: req.body } }); - return res.send(`Operation result for ${idShortPath}`).end(); + return res.send(success ? 204 : 500).end(); } case "/invoke-async": { if (req.method !== "POST") return res.status(405).end(); + const op = Traverser.getElementByIdPath(this.aas, sm, idShortPath); + if (op === null || !types.isOperation(op)) return res.status(404).end(); // TODO // Invoke operation and get handle + const request: Request = { type: "CALL-ASYNC", target: op, extraData: { args: req.body } }; - const handle = "TODO"; + const handle = this.onRequestCallback(request); const operationPaths = [`${idShortPath.join("/")}/operation-status/${handle}`, `${idShortPath.join("/")}/operation-result/${handle}`]; return res.status(202).header("Location", operationPaths).end(); @@ -171,17 +186,24 @@ export default class HTTPInterfaceServer extends AASInterfaceServer { switch (operation) { case "operation-status": { if (req.method !== "GET") return res.status(405).end(); - // TODO + + const state = this.onRequestCallback({ type: "GET-OP-STATE", target: handle }); + + if (state === undefined) return res.status(404).end(); + return res.json({ - "status": "TODO", + finished: state, handle }).end(); } case "operation-result": { if (req.method !== "GET") return res.status(405).end(); - // TODO + + const result = this.onRequestCallback({ type: "GET-OP-RESULT", target: handle }); + if (result === null) return res.status(404).end(); + return res.json({ - "status": "TODO", + result, handle }).end(); } diff --git a/lib/src/example_modules/mqttConnector.ts b/lib/src/example_modules/mqttConnector.ts index c2e1ed9..2a41cab 100644 --- a/lib/src/example_modules/mqttConnector.ts +++ b/lib/src/example_modules/mqttConnector.ts @@ -3,11 +3,11 @@ import type { Property, Operation, OperationVariable, BasicEventElement } from " import InterfaceConnectionObject from "interfaceConnectionObject"; import type { types } from "@aas-core-works/aas-core3.0-typescript"; -class MQTTConnector extends InterfaceConnectionObject { - protected name: string = "MQTT Connector"; - protected uriProtocol: string[] = ["mqtt", "mqtts"]; - protected connectionType: "ON_DEMAND" | "PERMANENT" = "PERMANENT"; - protected supportsSubscriptions: boolean = true; +export default class MQTTConnector extends InterfaceConnectionObject { + public name: string = "MQTT Connector"; + public uriProtocol: string[] = ["mqtt", "mqtts"]; + public connectionType: "ON_DEMAND" | "PERMANENT" = "PERMANENT"; + public supportsSubscriptions: boolean = true; private client: mqtt.MqttClient|null = null; @@ -79,21 +79,14 @@ class MQTTConnector extends InterfaceConnectionObject { return true; } - public callActionSync(action: types.Operation, args: OperationVariable[]): boolean { + public callActionSync(action: types.Operation, args: Record): any { // TODO // Vorher mal ne ordentliche Mapping-Definition throw new Error("Method not implemented."); } - public callActionAsync(action: types.Operation, args: OperationVariable[]): string | null { - // TODO - // Vorher mal ne ordentliche Mapping-Definition - - throw new Error("Method not implemented."); - } - - public readAsyncActionResponse(action: types.Operation): any { + public callActionAsync(action: types.Operation, args: Record): string | null { // TODO // Vorher mal ne ordentliche Mapping-Definition diff --git a/lib/src/interfaceConnectionObject.ts b/lib/src/interfaceConnectionObject.ts index cc6be36..57d615d 100644 --- a/lib/src/interfaceConnectionObject.ts +++ b/lib/src/interfaceConnectionObject.ts @@ -2,16 +2,17 @@ import type { types } from "@aas-core-works/aas-core3.0-typescript"; import { AIMCMapper } from "parser/AIMCMapper"; import { v4 } from "uuid"; import type { EndpointMetadata } from "types/aidConf"; +import { Request } from "types/requests"; type ConnectionType = "ON_DEMAND" | "PERMANENT"; -export type OnResponseCallback = (response: any) => void +export type OnEventCallback = (response: any) => void export default abstract class InterfaceConnectionObject { - protected abstract readonly name: string; - protected abstract readonly uriProtocol: string[]|string; - protected abstract readonly connectionType: ConnectionType; - protected abstract readonly supportsSubscriptions: boolean; + public abstract readonly name: string; + public abstract readonly uriProtocol: string[]|string; + public abstract readonly connectionType: ConnectionType; + public abstract readonly supportsSubscriptions: boolean; protected observerStore: Record void)[]> = {}; protected eventSubStore: Record void)[]> = {}; @@ -19,9 +20,9 @@ export default abstract class InterfaceConnectionObject { public constructor( protected readonly connectionParameter: ConfigInterface, - protected readonly endpointMetadata: EndpointMetadata, + public readonly endpointMetadata: EndpointMetadata, protected readonly mapper: AIMCMapper, - private readonly onResponseCallback: OnResponseCallback) {} + protected readonly onConnectorEvent: OnEventCallback) {} public abstract connect(): boolean; @@ -33,11 +34,17 @@ export default abstract class InterfaceConnectionObject { public abstract observeProperty(prop: types.Property, callback: (value: any) => void): boolean; - public abstract callActionSync(action: types.Operation, args: types.OperationVariable[]): boolean; + public abstract callActionSync(action: types.Operation, args: Record): any; - public abstract callActionAsync(action: types.Operation, args: types.OperationVariable[]): string | null; + public abstract callActionAsync(action: types.Operation, args: Record): string | null; - public abstract readAsyncActionResponse(action: types.Operation): any; + public readAsyncActionState(handle: string): any { + return this.asyncActionStateStore[handle].finished; + } + + public readAsyncActionResponse(handle: string): any { + return this.asyncActionStateStore[handle].result; + }; public abstract subscribeEvent(event: types.BasicEventElement, callback: (event: types.BasicEventElement) => void): boolean; diff --git a/lib/src/multimessageBroker.ts b/lib/src/multimessageBroker.ts index 5fb9f54..826f9a6 100644 --- a/lib/src/multimessageBroker.ts +++ b/lib/src/multimessageBroker.ts @@ -3,7 +3,7 @@ import type AASInterfaceServer from "server"; import type { Request } from "types/requests"; import type InterfaceConnectionObject from "interfaceConnectionObject"; import Traverser from "./helper/traverser"; -import type { AIMCMapper } from "./parser/AIMCMapper"; +import { AIMCMapper } from "./parser/AIMCMapper"; import { AssetInterfacesDescription } from "types/aidConf"; import AIDParser from "./parser/AIDParser"; @@ -13,9 +13,9 @@ type AASRegistration = { } type AASRegistrationPrepared = AASRegistration & { - serverInstaces?: AASInterfaceServer[]; - connectorInterfaces?: InterfaceConnectionObject[]; - mappingConfiguration?: AIMCMapper; + serverInstances: AASInterfaceServer[]; + connectorInterfaces: InterfaceConnectionObject[]; + mappingConfiguration: AIMCMapper; } type ServerInterfaceEntry = { @@ -23,15 +23,30 @@ type ServerInterfaceEntry = { config: ConfigInterface } +type InterfaceConnectionEntry = { + interfaceConnection: typeof InterfaceConnectionObject, + config: ConfigInterface +} + export default class MultiMessageBroker { private prepared: boolean = false; private aasRegistrations: AASRegistrationPrepared[] = []; + private interfaceConnections: InterfaceConnectionEntry[] = []; public constructor() {} - public registerAAS(registration: AASRegistration): void { - this.aasRegistrations.push(registration as AASRegistrationPrepared); + public registerAAS(registration: AASRegistration): void { + this.aasRegistrations.push({ + ...registration, + serverInstances: [], + connectorInterfaces: [], + mappingConfiguration: new AIMCMapper(registration.aas) + }); + } + + public registerInterfaceConnection(interfaceConnectionEntry: InterfaceConnectionEntry): void { + this.interfaceConnections.push(interfaceConnectionEntry); } public prepare(): void { @@ -39,10 +54,10 @@ export default class MultiMessageBroker { if (!Array.isArray(registration.serverInterfaces)) registration.serverInterfaces = [registration.serverInterfaces]; - registration.serverInstaces = registration.serverInterfaces.map(serverInterface => { + registration.serverInstances = registration.serverInterfaces.map(serverInterface => { try { // @ts-ignore - const server = new serverInterface.serverInterface(serverInterface.config, registration.aas, this.onInterfaceRequest.bind(this)); + const server = new serverInterface.serverInterface(serverInterface.config, registration.aas, registration.mappingConfiguration, (req: Request) => this.onInterfaceRequest(req, registration)); server.prepare(); return server; } catch (e) { @@ -52,9 +67,17 @@ export default class MultiMessageBroker { }).filter(i => i !== null); const interfaceDescription = AIDParser.parse(registration.aas); + if (!interfaceDescription) continue; - // TODO - // Create connectors and mapping configuration + const endpoints = Object.values(interfaceDescription).flatMap(idEntries => idEntries.map(entry => entry.EndPointMetadata)); + const uniqueProtocols = [...new Set(endpoints.map(endpoint => new URL(endpoint.base).protocol))]; + + for (const protocol of uniqueProtocols) { + const connectorProto = this.interfaceConnections.find(connection => connection.interfaceConnection.prototype.uriProtocol.includes(protocol)) + if (connectorProto === undefined) continue; + // @ts-ignore + registration.connectorInterfaces.push(new connectorProto.interfaceConnection(connectorProto.config, registration.mappingConfiguration, (response: any) => this.onConnectorEvent(response))); + } } this.prepared = true; @@ -64,17 +87,74 @@ export default class MultiMessageBroker { if (!this.prepared) this.prepare(); for (const registration of this.aasRegistrations) { - registration.serverInstaces?.forEach(server => server?.run()); + registration.serverInstances?.forEach(server => server?.run()); } } - private onInterfaceRequest(request: Request) { - // TODO + private onInterfaceRequest(request: Request, registration: AASRegistrationPrepared): any { + + const getConnector = (target: types.Class) => { + const mapping = registration.mappingConfiguration?.get(target); + if (mapping === undefined) throw new ReferenceError(`No mapping found for ${(request.target as any).idShort}`); + + const connector = registration.connectorInterfaces?.find(connector => connector.endpointMetadata.base === mapping.base); + if (connector === undefined) throw new ReferenceError(`No connector found for ${(request.target as any).idShort}`); + + return connector; + } + + switch (request.type) { + case "READ": { + const connector = getConnector(request.target); + + return connector.readProperty(request.target); + } + case "WRITE": { + const connector = getConnector(request.target); + + return connector.writeProperty(request.target, request.extraData.value); + } + case "OBSERVE": { + const connector = getConnector(request.target); + + return connector.observeProperty(request.target, request.extraData.callback); + } + case "SUBSCRIBE": { + const connector = getConnector(request.target); + + return connector.subscribeEvent(request.target, request.extraData.callback); + } + case "UNSUBSCRIBE": { + const connector = getConnector(request.target); + + return connector.unsubscribeEvent(request.target); + } + case "CALL": { + const connector = getConnector(request.target); + + return connector.callActionSync(request.target, request.extraData.args); + } + case "CALL-ASYNC": { + const connector = getConnector(request.target); + + return connector.callActionAsync(request.target, request.extraData.args); + } + case "GET-OP-STATE": { + const state = registration.connectorInterfaces.map(connector => connector.readAsyncActionState(request.target)).filter(state => state !== undefined)[0]; + if (state === undefined) throw new ReferenceError(`No connector found for operation handle ${request.target}`); + return state; + } + case "GET-OP-RESULT": { + const result = registration.connectorInterfaces.map(connector => connector.readAsyncActionResponse(request.target)).filter(result => result !== undefined)[0]; + if (result === undefined) throw new ReferenceError(`No connector found for operation handle ${request.target}`); + return result; + } + } } private onConnectorEvent(response: any) { for (const aas of this.aasRegistrations) { - for (const server of aas.serverInstaces ?? []) { + for (const server of aas.serverInstances ?? []) { server?.notify(response); } } diff --git a/lib/src/server.ts b/lib/src/server.ts index 2acfef4..dafb0fc 100644 --- a/lib/src/server.ts +++ b/lib/src/server.ts @@ -6,7 +6,7 @@ export default abstract class AASInterfaceServer { protected config: ConfigInterface; protected abstract readonly supportsSubscriptions: boolean; protected readonly aas: types.Environment; - private onRequestCallback: OnRequestCallback + protected onRequestCallback: OnRequestCallback constructor(config: ConfigInterface, aas: types.Environment, onRequestCallback: OnRequestCallback) { this.config = config; diff --git a/lib/src/types/aidConf.ts b/lib/src/types/aidConf.ts index 0fd046b..ab1e3f5 100644 --- a/lib/src/types/aidConf.ts +++ b/lib/src/types/aidConf.ts @@ -1,7 +1,7 @@ import { AvailableEndpoint } from "./common"; export type AssetInterfacesDescription = { - [key in AvailableEndpoint]: InterfaceDescription[]; + [key in AvailableEndpoint]?: InterfaceDescription[]; }; export type InterfaceDescription = { diff --git a/lib/src/types/requests.ts b/lib/src/types/requests.ts index 4155ccd..0110ad5 100644 --- a/lib/src/types/requests.ts +++ b/lib/src/types/requests.ts @@ -1,11 +1,85 @@ -import { types } from "@aas-core-works/aas-core3.0-typescript"; +import type { types } from "@aas-core-works/aas-core3.0-typescript"; -export type Request = { - type: RequestType; - target: types.Reference | types.Class | string[]; - extraData?: any; +export type Request = GetRequest | WriteRequest | ObserveRequest | CallRequest | CallAsyncRequest | AsyncStateRequest | AsyncResultRequest | SubscribeRequest | UnsubscribeRequest; + +export type OnRequestCallback = (request: Request) => any & GetRequestCallback & WriteRequestCallback & ObserveRequestCallback & CallRequestCallback & CallAsyncRequestCallback & AsyncStateRequestCallback & AsyncResultRequestCallback & SubscribeRequestCallback & UnsubscribeRequestCallback; + +type GetRequest = { + type: "READ"; + target: types.Property; } -export type OnRequestCallback = (request: Request) => void +type GetRequestCallback = (request: GetRequest) => any; + +type WriteRequest = { + type: "WRITE"; + target: types.Property; + extraData: { + value: any; + } +} + +type WriteRequestCallback = (request: WriteRequest) => void; + +type ObserveRequest = { + type: "OBSERVE"; + target: types.Property; + extraData: { + callback: (value: any) => void; + } +} + +type ObserveRequestCallback = (request: ObserveRequest) => boolean; + +type SubscribeRequest = { + type: "SUBSCRIBE"; + target: types.BasicEventElement; + extraData: { + callback: (value: any) => void; + } +} + +type SubscribeRequestCallback = (request: SubscribeRequest) => boolean; + +type UnsubscribeRequest = { + type: "UNSUBSCRIBE"; + target: types.BasicEventElement; +} + +type UnsubscribeRequestCallback = (request: UnsubscribeRequest) => void; + +type CallRequest = { + type: "CALL"; + target: types.Operation; + extraData: { + args: Record; + } +} + +type CallRequestCallback = (request: CallRequest) => any; + +type CallAsyncRequest = { + type: "CALL-ASYNC"; + target: types.Operation; + extraData: { + args: Record; + } +} + +type CallAsyncRequestCallback = (request: CallAsyncRequest) => string | null; + +type AsyncStateRequest = { + type: "GET-OP-STATE"; + target: string; +} + +type AsyncStateRequestCallback = (request: AsyncStateRequest) => boolean; + +type AsyncResultRequest = { + type: "GET-OP-RESULT"; + target: string; +} + +type AsyncResultRequestCallback = (request: AsyncResultRequest) => any; type RequestType = "READ" | "WRITE" | "OBSERVE" | "CALL" | "CALL-ASYNC" | "GET-OP-STATE" | "GET-OP-RESULT" | "SUBSCRIBE" | "UNSUBSCRIBE" \ No newline at end of file