import { types } from "@aas-core-works/aas-core3.0-typescript"; import type AbstractInterfaceServer from "./abstractInterfaceServer"; import type { Request } from "./types/requests"; import type AbstractConnectionObject from "./abstractConnectionObject"; import AIMCMapper from "./AIMCMapper"; import AIDParser from "./parser/AIDParser"; import type { EndpointMetadata, InterfaceActionForm, InterfaceEventForm, InterfaceForm, InterfacePropertyForm } from "./types/aidConf"; import type { ConnectionConfiguration } from "./types/requests"; type AASRegistration = { aas: types.Environment, serverInterfaces: ServerInterfaceEntry | ServerInterfaceEntry[], } type AASRegistrationPrepared = AASRegistration & { serverInstances: AbstractInterfaceServer[]; connectorInterfaces: AbstractConnectionObject[]; mappingConfiguration: AIMCMapper; } type ServerInterfaceEntry = { serverInterface: typeof AbstractInterfaceServer, config: ConfigInterface } type InterfaceConnectionEntry = { interfaceConnection: typeof AbstractConnectionObject, config: ConfigInterface } /** * The core of the library. * @remarks * Here are all Interface Servers and Connectors are created and managed. * Also every request is handled here. * @public */ export default class MultiMessageBroker { /** * MMB singleton instance * @private */ private static instance: MultiMessageBroker|null = null; /** * Whether the broker is prepared. * @private */ private prepared: boolean = false; /** * All registered AASs. * @private */ private aasRegistrations: AASRegistrationPrepared[] = []; /** * All registered Interface Connectors. * @private */ private interfaceConnections: InterfaceConnectionEntry[] = []; /** * @private */ private constructor() {} /** * Get the singleton instance of the broker. * @public */ public static getInstance(): MultiMessageBroker { if (this.instance === null) this.instance = new MultiMessageBroker(); return this.instance; } /** * Register an AAS. * @remarks * This will also create the {@link AIMCMapper} for the AAS * @param registration AAS registration */ public registerAAS(registration: AASRegistration): void { this.aasRegistrations.push({ ...registration, serverInstances: [], connectorInterfaces: [], mappingConfiguration: new AIMCMapper(registration.aas) }); } /** * Register an Interface Connector. * @param interfaceConnectionEntry Interface connector registration * @typeParam T - The config interface for your interface connector. */ public registerInterfaceConnection(interfaceConnectionEntry: InterfaceConnectionEntry): void { this.interfaceConnections.push(interfaceConnectionEntry); } /** * Prepare the broker. * @remarks * This will create instances of the {@link AbstractInterfaceServer} and {@link AbstractConnectionObject} classes. */ public prepare(): void { for (const registration of this.aasRegistrations) { if (!Array.isArray(registration.serverInterfaces)) registration.serverInterfaces = [registration.serverInterfaces]; registration.serverInstances = registration.serverInterfaces.map(serverInterface => { try { // @ts-ignore const server = new serverInterface.serverInterface(serverInterface.config, registration.aas, (req: Request) => this.onInterfaceRequest(req, registration)); server.prepare(); return server; } catch (e) { console.error("Error while creating server interface instance", e); return null; } }).filter(i => i !== null); const interfaceDescription = AIDParser.parse(registration.aas); if (!interfaceDescription) continue; const endpoints = Object.values(interfaceDescription).flatMap(idEntries => idEntries.map(entry => entry.EndPointMetadata)); const uniqueProtocols: Record = {}; endpoints.forEach(endpoint => { const pUrl = new URL(endpoint.base); if (!uniqueProtocols[pUrl.protocol]) uniqueProtocols[pUrl.protocol] = [endpoint]; else uniqueProtocols[pUrl.protocol].push(endpoint); }); for (const [protocol, endpoints] of Object.entries(uniqueProtocols)) { const connectorProto = this.interfaceConnections.find(connection => { const protos = typeof connection.interfaceConnection.uriProtocol === "string" ? [connection.interfaceConnection.uriProtocol] : connection.interfaceConnection.uriProtocol; return protos.map(proto => proto.endsWith(":") ? proto : proto + ":").includes(protocol.endsWith(":") ? protocol : protocol + ":"); }); if (connectorProto === undefined) continue; // @ts-ignore for (const ep of endpoints) registration.connectorInterfaces.push(new connectorProto.interfaceConnection(connectorProto.config, ep, registration.mappingConfiguration, (response: any) => this.onConnectorEvent(response))); } } this.prepared = true; } /** * Start the broker. */ public start(): void { if (!this.prepared) this.prepare(); for (const registration of this.aasRegistrations) { registration.connectorInterfaces.forEach(connector => connector.connect()); registration.serverInstances?.forEach(server => server?.run()); } } /** * Callback on an server request * @param request Request * @param registration AAS Registration * @returns Value or success state or nothing based on the request type */ private onInterfaceRequest(request: Request, registration: AASRegistrationPrepared): any { const getMapping = (target: types.Class) => { const mapping = registration.mappingConfiguration?.get(target); if (mapping === undefined) throw new ReferenceError(`No mapping found for ${(request.target as any).idShort}`); return mapping; } const getConnector = (mapping: ConnectionConfiguration) => { 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; } if (request.type === "xAsyncActionState" || request.type === "xAsyncActionResult") { if (request.type === "xAsyncActionState") { 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; } else { 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; } } else { const mapping = getMapping(request.target); mapping.forms = mapping.forms.filter((form: InterfaceForm) => form.op.toLocaleLowerCase() === request.type.toLocaleLowerCase()); if (mapping.forms.length === 0) throw new ReferenceError(`No form found for ${(request.target as any).idShort} with operation ${request.type}`); const connector = getConnector(mapping); switch (request.type) { case "readProperty": return connector.readProperty(request.target, mapping); case "writeProperty": return connector.writeProperty(request.target, mapping, request.extraData.value); case "observeProperty": return connector.observeProperty(request.target, mapping, request.extraData.callback); case "unobserveProperty": return connector.unobserveProperty(request.target, mapping); case "subscribeEvent": { return connector.subscribeEvent(request.target, mapping, request.extraData.callback); } case "unsubscribeEvent": { return connector.unsubscribeEvent(request.target, mapping); } case "invokeAction": return request.extraData.async ? connector.callActionAsync(request.target, mapping, request.extraData.args) : connector.callActionSync(request.target, mapping, request.extraData.args); } } } /** * Callback on a connector event. * @param response * @alpha */ private onConnectorEvent(response: any) { for (const aas of this.aasRegistrations) { for (const server of aas.serverInstances ?? []) { server?.notify(response); } } } }