diplomarbeit-code/lib/src/multimessageBroker.ts

import { types } from "@aas-core-works/aas-core3.0-typescript";
import type AASInterfaceServer from "./server";
import type { Request } from "./types/requests";
import type InterfaceConnectionObject from "interfaceConnectionObject";
import AIMCMapper from "./parser/AIMCMapper";
import AIDParser from "./parser/AIDParser";
import { EndpointMetadata } from "types/aidConf";

type AASRegistration = {
    aas: types.Environment,
    serverInterfaces: ServerInterfaceEntry<any> | ServerInterfaceEntry<any>[],
}

type AASRegistrationPrepared = AASRegistration & {
    serverInstances: AASInterfaceServer<any>[];
    connectorInterfaces: InterfaceConnectionObject<any>[];
    mappingConfiguration: AIMCMapper;
}

type ServerInterfaceEntry<ConfigInterface> = {
    serverInterface: typeof AASInterfaceServer<ConfigInterface>,
    config: ConfigInterface
}

type InterfaceConnectionEntry<ConfigInterface> = {
    interfaceConnection: typeof InterfaceConnectionObject<ConfigInterface>,
    config: ConfigInterface
}

export default class MultiMessageBroker {
    private prepared: boolean = false;

    private aasRegistrations: AASRegistrationPrepared[] = [];
    private interfaceConnections: InterfaceConnectionEntry<any>[] = [];

    public constructor() {}

    public registerAAS<T>(registration: AASRegistration): void {
        this.aasRegistrations.push({
            ...registration,
            serverInstances: [],
            connectorInterfaces: [],
            mappingConfiguration: new AIMCMapper(registration.aas)
        });
    }

    public registerInterfaceConnection<T>(interfaceConnectionEntry: InterfaceConnectionEntry<T>): void {
        this.interfaceConnections.push(interfaceConnectionEntry);
    }

    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<typeof serverInterface.config>(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<string, EndpointMetadata[]> = {};
            
            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;
    }

    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());
        }
    }

    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.serverInstances ?? []) {
                server?.notify(response);
            }
        }
    }
}