communication-binding.ts

/*! Copyright (c) 2020 Siemens AG. Licensed under the MIT License. */

import { EventEmitter } from "events";

/**
 * Indicates the connectivity state of a communication binding with respect to
 * the Coaty communication infrastructure.
 *
 * @remarks For bindings that use brokerless peer-to-peer messaging protocols
 * the value emitted while the binding is joined is `Online`, if and only if the
 * agent is not isolated, i.e. can communicate with other agents.
 */
export enum CommunicationState {

    /** Disconnected from the Coaty communication infrastructure. */
    Offline,

    /** Connected to the Coaty communication infrastructure. */
    Online,
}

/**
 * Predefined event types for Coaty communication event patterns.
 */
export enum CommunicationEventType {
    // Event types for non-Coaty messages
    External = 0,

    Raw,

    // Event types for Coaty one-way messages
    OneWay,

    Advertise,
    Deadvertise,
    Channel,
    Associate,
    IoValue,

    // Event types for Coaty two-way messages
    TwoWay,

    Discover,
    Resolve,
    Query,
    Retrieve,
    Update,
    Complete,
    Call,
    Return,
}

/**
 * Protocol for exchanging communication event properties for both publications and
 * subscriptions between a binding and the communication manager.
 */
export interface CommunicationEventLike {

    /** Event type. */
    eventType: CommunicationEventType;

    /** 
     * Event type filter (for event types with filter only).
     *
     * @remarks For Raw events this property specifies the publication or
     * subscription topic in a binding-specific format.
     *
     * @remarks For IoValue events this property specifies the associating IO
     * route in a binding-specific.
     */
    eventTypeFilter?: string;

    /** 
     * Object ID of the event source object (except for Raw and external IoValue
     * events).
     */
    sourceId?: string;

    /** 
     * Correlation ID for inbound/outbound two-way request-response events.
     * Correlated subscription topic for inbound Raw and IoValue events.
     */
    correlationId?: string;

    /** Event data (for publication events only). */
    data?: any;

    /** 
     * Indicates whether data property contains raw data (string or Uint8Array
     * or Buffer) or a JSON object (used for Raw and IO value publication and
     * subscription events only).
     *
     * If not specified, the value defaults to false.
     */
    isDataRaw?: boolean;

    /** 
     * Indicates whether the route of an IO value subscription event is external
     * or not (used for IO value subscription events only).
     *
     * If not specified, the value defaults to false.
     */
    isExternalIoRoute?: boolean;

    /** 
     * Binding-specific options for publication, subscription, or unsubscription.
     * (optional)
     */
    options?: { [key: string]: any };
}

/**
 * Defines static constructor signature for communication binding types that
 * extend and implement the abstract `CommunicationBinding` class.
 */
export type CommunicationBindingStatic<T extends CommunicationBinding<O>, O extends CommunicationBindingOptions> = new (options: O) => T;

/**
 * A utility type used to specify a value for the `CommunicationOptions.binding`
 * property in the agent container configuration.
 */
export type CommunicationBindingWithOptions<T extends CommunicationBinding<O>, O extends CommunicationBindingOptions> = {
    type: CommunicationBindingStatic<T, O>,
    options: O,
};

/**
 * Defines log levels for logging within a communication binding.
 */
export enum CommunicationBindingLogLevel {

    /** Log debug messages. */
    debug = 0,

    /** Log informational messages. */
    info = 1,

    /** Log error messages. */
    error = 2,
}

/**
 * Defines common options of all communication binding types.
 */
export interface CommunicationBindingOptions {

    /**
     * Namespace used to isolate different Coaty applications running with the
     * same binding type in the same networking infrastructure (optional).
     *
     * Communication events are only routed between agents within a common
     * communication namespace: an agent publishes and observes communication
     * events with respect to the given namespace.
     *
     * If not specified (i.e. undefined), a default namespace named "-" is used.
     */
    namespace?: string;

    /**
     * Determines whether to enable cross-namespace communication between agents
     * in special use cases (optional). 
     *
     * If `true`, an agent receives communication events published by *any*
     * agent running with the same binding type in the same networking
     * infrastructure, regardless of namespace.
     *
     * If not specified or `false`, this option is not enabled.
     */
    shouldEnableCrossNamespacing?: boolean;

    /**
     * The level used to log errors, informational, as well as debug messages
     * generated by the binding (optional).
     *
     * Messages corresponding to the specified log level are emitted on the
     * "error", "info", or "debug" event, respectively. If level is "error",
     * only errors are emitted; if level is "info", only "error" and "info"
     * messages are emitted; if level is "debug", all messages are emitted.
     *
     * If not specified, the log level defaults to "error".
     */
    logLevel?: CommunicationBindingLogLevel;
}

/**
 * Defines the three lifecycle states of a communication binding.
 *
 * The lifecycle of a binding transitions from "Initialized" to "Joined" to
 * "Unjoined". An unjoined binding cannot be joined or initialized again.
 */
export enum CommunicationBindingState {

    /** Binding is initialized. */
    Initialized,

    /** Binding has joined the Coaty communication infrastructure. */
    Joined,

    /** Binding has left the Coaty communication infrastructure. */
    Unjoined,
}

/**
 * Defines options for joining the Coaty communication infrastructure.
 */
export interface CommunicationBindingJoinOptions {

    /** 
     * An id that uniquely identifies the joining agent. 
     *
     * A binding might use this id e.g. as a client or session id for the
     * underlying messaging transport.
     */
    agentId: string;

    /**
     * One-way communication event likes to be published to subscribing agents
     * whenever this binding joins or rejoins the Coaty communication
     * infrastructure.
     *
     * On joining, these events must always be published first, before any other
     * pending events.
     *
     * @remarks These events **must** also be republished after temporary
     * disconnections in a broker-based communication infrastructure.
     */
    joinEvents: CommunicationEventLike[];

    /**
     * A one-way communication event like that **must** be delivered to subscribing
     * agents whenever this binding unjoins the Coaty communication infrastructure
     * expectedly or unexpectedly.
     *
     * @remarks The event must not only be delivered when the `unjoin` method is
     * called; it must also be delivered whenever this binding (temporarily) becomes
     * offline, e.g. if the underlying messaging transport is destroyed abnormally
     * or if the agent process is terminated abnormally.
     */
    unjoinEvent: CommunicationEventLike;
}

/**
 * Defines a generic public API for transmitting Coaty communication events via
 * specific broker-based or brokerless publish-subscribe messaging protocols.
 */
export interface CommunicationBindingProtocol {

    /* Variables */

    /**
     * Gets the current lifecycle state of this communication binding.
     */
    readonly state: CommunicationBindingState;

    /**
     * Defines the name of this binding used for identification and display
     * purposes.
     */
    readonly apiName: string;

    /**
     * Defines the protocol version number of this binding, a positive integer.
     *
     * Increment this version whenever you change or add new communication
     * events or make breaking changes to the binding's messaging protocol.
     *
     * Note that each specific binding type provides its own API versioning. API
     * versions are not synchronized or shared across different binding types.
     */
    readonly apiVersion: number;

    /* Event Emitters */

    /** 
     * Emitted whenever this binding signals an error, informational or debug
     * message.
     *
     * @remarks This event should be used in binding classes to log changes to
     * connection state ("info"), errors ("error"), as well as publications and
     * subscriptions/unsubscriptions ("debug").
     */
    on(event: "error" | "info" | "debug", listener: (msg: string) => void): void;

    /**
     * Emitted whenever the binding's communication state changes.
     *
     * @remarks For bindings that use brokerless peer-to-peer messaging
     * protocols the value emitted while the binding is joined is `Online`, if
     * and only if the agent is not isolated, i.e. can communicate with other
     * agents.
     */
    on(event: "communicationState", listener: (state: CommunicationState) => void): void;

    /**
     * Emitted whenever an inbound communication event has arrived.
     *
     * For Raw events, the `eventTypeFilter` property contains the published
     * topic name and the `correlationId` property contains the correlated
     * subscription topic.
     *
     * For IoValue events, the `eventTypeFilter` property contains the IO route.
     */
    on(event: "inboundEvent", listener: (eventLike: CommunicationEventLike) => void): void;

    /* Methods */

    /**
     * Called by CommunicationManager once when this binding should join the
     * Coaty communication infrastructure with the given options.
     *
     * @param joinOptions options used for joining
     */
    join(joinOptions: CommunicationBindingJoinOptions): void;

    /**
     * Called by CommunicationManager once when this binding should leave the
     * Coaty communication infrastructure and be disposed.
     *
     * The promise returned should always be resolved when the binding has
     * completed unjoining so that its safe to invoke another `join()` operation
     * on a *new* instance of this binding type.
     *
     * When unjoining, the binding instance should release all internally used
     * resources so that it may be garbage collected. No further API method
     * calls will be invoked on this binding.
     *
     * @returns a Promise that is always resolved when the binding has completed
     * unjoining.
     */
    unjoin(): Promise<void>;

    /**
     * Called by CommunicationManger to publish the given event.
     *
     * Publications issued before the binding has joined are deferred.
     * Publications issued after the binding has unjoined are silently
     * discarded.
     *
     * @remarks Each binding must support publishing of all Coaty communication
     * events except Raw and external IoValue events. If a binding doesn't
     * support these, they are discarded and an error is emitted on the "error"
     * event emitter.
     *
     * @remarks A Raw event that specifies a publication topic with a Coaty-like
     * shape is discarded and an error is emitted on the "error"  event emitter.
     *
     * @param eventLike represents the event to be published
     */
    publish(eventLike: CommunicationEventLike): void;

    /**
     * Called by CommunicationManger to subscribe the given event.
     *
     * Subscriptions issued before the binding has joined are deferred.
     * Subscriptions issued after the binding has unjoined are silently
     * discarded.
     *
     * @remarks Each binding must support subscribing all Coaty communication
     * events except Raw and external IoValue events. If a binding doesn't
     * support these, they are discarded and an error is emitted on the "error"
     * event emitter.
     *
     * @remarks A Raw event that specifies a subscription topic with a
     * Coaty-like shape is discarded and an error is emitted on the "error"
     * event emitter.
     *
     * @param eventLike represents the event to be subscribed
     */
    subscribe(eventLike: CommunicationEventLike): void;

    /**
     * Called by CommunicationManger to unsubscribe the given event.
     *
     * Unsubscriptions issued before the binding has joined are ignored.
     * Unsubscriptions issued after the binding has unjoined are silently
     * discarded.
     *
     * @remarks Each binding must support unsubscribing all Coaty communication
     * events except Raw and external IoValue events. If a binding doesn't
     * support these, they are discarded and an error is emitted on the "error"
     * event emitter.
     *
     * @param eventLike represents the event to be unsubscribed
     */
    unsubscribe(eventLike: CommunicationEventLike): void;

    /**
     * Create an IO route for routing IO values of the given IO source to associated
     * IO actors.
     *
     * @param ioSourceId object ID of an IoSource
     * @returns an associating topic for routing
     */
    createIoRoute(ioSourceId: string): string;
}

/**
 * An abstract class that provides the generic communication binding protocol and
 * additional abstract methods to be implemented by concrete implementation
 * subclasses.
 *
 * @remarks The generic type argument `O` specifies the concrete object interface of
 * the binding's options.
 */
export abstract class CommunicationBinding<O extends CommunicationBindingOptions>
    extends EventEmitter implements CommunicationBindingProtocol {

    protected static readonly DEFAULT_NAMESPACE = "-";

    private _state: CommunicationBindingState;

    /**
     * Gets the binding options (read-only).
     *
     * @remarks For the optional base binding options defined in
     * `CommunicationBindingOptions`, (default) values are always set by the
     * base binding class. Thus, a concrete binding class need not check whether
     * a value is defined for these properties.
     */
    get options(): Readonly<O> {
        return this._options;
    }

    get state() {
        return this._state;
    }

    abstract get apiName(): string;

    abstract get apiVersion(): number;

    /**
     * @internal For internal use in framework only.
     *
     * Create an instance of this binding type with the given options.
     *
     * @param _options the non-optional binding options
     */
    constructor(private _options: O) {
        super();
        this._state = CommunicationBindingState.Initialized;
        if (this._options.logLevel === undefined) {
            this._options.logLevel = CommunicationBindingLogLevel.error;
        }
        if (this._options.namespace === undefined) {
            this._options.namespace = CommunicationBinding.DEFAULT_NAMESPACE;
        }
        if (this._options.shouldEnableCrossNamespacing === undefined) {
            this._options.shouldEnableCrossNamespacing = false;
        }
        this.onInit();
    }

    join(joinOptions: CommunicationBindingJoinOptions) {
        if (this.state === CommunicationBindingState.Joined) {
            return;
        }
        if (this.state === CommunicationBindingState.Unjoined) {
            this.log(CommunicationBindingLogLevel.error, "Cannot join binding which is in Unjoined state");
            return;
        }
        this._state = CommunicationBindingState.Joined;
        this.onJoin(joinOptions);
    }

    unjoin() {
        if (this.state === CommunicationBindingState.Unjoined) {
            return Promise.resolve();
        }
        this._state = CommunicationBindingState.Unjoined;
        return this.onUnjoin().then(() => { this.removeAllListeners(); });
    }

    publish(eventLike: CommunicationEventLike) {
        if (this.state === CommunicationBindingState.Unjoined) {
            return;
        }
        this.onPublish(eventLike);
    }

    subscribe(eventLike: CommunicationEventLike) {
        if (this.state === CommunicationBindingState.Unjoined) {
            return;
        }
        this.onSubscribe(eventLike);
    }

    unsubscribe(eventLike: CommunicationEventLike) {
        if (this.state === CommunicationBindingState.Unjoined) {
            return;
        }
        this.onUnsubscribe(eventLike);
    }

    abstract createIoRoute(ioSourceId: string): string;

    /* Communication Binding Protocol Handlers */

    /** 
     * Called by base binding constructor once to initialize subclass members.
     */
    protected abstract onInit(): void;

    /** 
     * Called by base binding class once to join.
     */
    protected abstract onJoin(joinOptions: CommunicationBindingJoinOptions): void;

    /** 
     * Called by base binding class once to unjoin.
     *
     * @remarks This method is also called if the binding is currently in
     * Initialized state. In this case, resources should be disposed.
     */
    protected abstract onUnjoin(): Promise<void>;

    /** 
     * Called by base binding class to publish the given event like.
     */
    protected abstract onPublish(eventLike: CommunicationEventLike): void;

    /** 
     * Called by base binding class to subscribe to the given event like.
     */
    protected abstract onSubscribe(eventLike: CommunicationEventLike): void;

    /** 
     * Called by base binding class to unsubscribe to the given event like.
     */
    protected abstract onUnsubscribe(eventLike: CommunicationEventLike): void;

    /* Utility Methods for Binding Subclasses */

    /**
     * Used to conditionally emit log messages on the "error", "info", or
     * "debug" event emitter, depending on the given log level and the target
     * log level specified in the binding options.
     *
     * If the target log level is "error", only error logs are emitted; if
     * target log level is "info", only "error" and "info" logs are emitted; if
     * target log level is "debug", all logs are emitted.
     *
     * If the given log level matches the binding's target log level, a message
     * formed by concatenating the partial arguments is emitted on the
     * corresponding event emitter.
     *
     * If any of the optional arguments is undefined, it is ignored in the
     * composed message.
     *
     * @param logLevel the log level
     * @param arg1 partial log message
     * @param arg2 partial log message (optional)
     * @param arg3 partial log message (optional)
     * @param arg4 partial log message (optional)
     * @param arg5 partial log message (optional)
     */
    protected log(logLevel: CommunicationBindingLogLevel, arg1: string, arg2?: any, arg3?: any, arg4?: any, arg5?: any) {
        const targetLogLevel = this.options.logLevel ?? CommunicationBindingLogLevel.error;
        if (targetLogLevel <= logLevel) {
            const event = CommunicationBindingLogLevel[logLevel];
            this.emit(event, (arg1 ?? "").concat(arg2 ?? "", arg3 ?? "", arg4 ?? "", arg5 ?? ""));
        }
    }

}