Adapter Interface

[aidanm3341]

To improve interoperability of the message broker with other messaging systems, we should add an interface for plugging in adapters which map messagebroker messages into messages of other systems in a way which is transparent to the publisher/subscriber of those messages.

The proposal is to introduce a new IMessageBrokerAdapter interface which can be given a configuration object to specify what channels should be exposed or consumed via the adapter. This adapter is then registered with an instance of the messagebroker and used behind the scenes for all relevant messages that are sent through that instance of the broker.

export interface IMessageBrokerAdapter<T> {
    readonly id: string;
    connect(): Promise<void>;
    disconnect(): Promise<void>;
    sendMessage(channelName: keyof T, message: IMessage): Promise<void>;
    subscribeToMessages(channelName: keyof T): Observable<IMessage>;
    isConnected(): boolean;
}

Here's a diagram to illustrate the separation of the application from the external messaging system.

graph TD
      App[Application] --> MB[MessageBroker]
      MB --> Channel[Internal Channels]
      MB --> Registry[Adapter Registry]

      Registry --> Adapter[IMessageBrokerAdapter]

      Adapter --> |sendMessage| External[External System]
      External --> |subscribeToMessages| Adapter
      Adapter --> |publishes to| Channel
      Channel --> |sends via| Adapter

      classDef interface fill:#e3f2fd,color:#000
      classDef core fill:#f3e5f5,color:#000
      classDef external fill:#fff3e0,color:#000

      class Adapter interface
      class App,MB,Channel,Registry core
      class External external

Loading

FDC3 Example:

const broker = messagebroker<IChannels>();

const fdc3Adapter: IMessageBrokerAdapter<IChannels> = new FDC3Adapter<IChannels>();

await fdc3Adapter.connect();

broker.registerAdapter(fdc3Adapter);

With the initial config done, the app can interact with the message broker with no knowledge of how the fdc3 communication is done.

broker.get('instrument').subscribe((message) => {
    console.log('Received instrument update:', message.data);
});

// publishes intent
broker.create('chart-request').publish({
    type: 'fdc3.instrument',
    name: 'Microsoft Corp',
    id: {
        ticker: 'MSFT'
    },
    timeframe: '1D'
});

// internal only
broker.create('price-updates').publish({
    ticker: 'AAPL',
    price: 151.20,
    timestamp: Date.now(),
    volume: 100000 
});

[aidanm3341]

I got AI to take a crack at implementing the adapter for FDC3 to help illustrate how it might work. It's not using all the best fdc3 tech like the actual library that it has, so while I'm sure this can be improved, it still illustrates the point.

import { Observable, Subject } from 'rxjs';
import { IMessage, IMessageBrokerAdapter } from '../contracts/contracts';
import { v4 as uuid } from 'uuid';

interface FDC3Context {
    type: string;
    id?: { [key: string]: string };
    name?: string;
    [key: string]: any;
}


interface FDC3DesktopAgent {
    addContextListener(handler: (context: FDC3Context) => void): Promise<void>;
    addIntentListener(intent: string, handler: (context: FDC3Context) => void): Promise<void>;
    broadcast(context: FDC3Context): Promise<void>;
    raiseIntent(intent: string, context?: FDC3Context): Promise<void>;
}

declare global {
    interface Window {
        fdc3?: FDC3DesktopAgent;
    }
}

export class FDC3Adapter<T> implements IMessageBrokerAdapter<T> {
    public readonly id: string = 'fdc3-adapter';
    public readonly name: string = 'FDC3 Desktop Agent Adapter';
    
    private connected: boolean = false;
    private fdc3: FDC3DesktopAgent | null = null;
    private messageSubjects: Map<string, Subject<IMessage>> = new Map();

    async connect(): Promise<void> {
        if (typeof window !== 'undefined' && window.fdc3) {
            this.fdc3 = window.fdc3;
        } else {
            throw new Error('FDC3 Desktop Agent not available');
        }

        try {
            await this.setupFDC3Listeners();
            this.connected = true;
        } catch (error) {
            throw new Error(`Failed to connect to FDC3: ${error}`);
        }
    }

    async disconnect(): Promise<void> {
        this.connected = false;
        this.messageSubjects.clear();
        this.fdc3 = null;
    }

    async sendMessage(channelName: string, message: IMessage): Promise<void> {
        if (!this.connected || !this.fdc3) {
            throw new Error('Adapter not connected');
        }

        try {
            if (channelName.startsWith('intent:')) {
                const intentName = channelName.replace('intent:', '');
                const fdc3Context = this.transformToFDC3Context(message);
                await this.fdc3.raiseIntent(intentName, fdc3Context);
            } else {
                const fdc3Context = this.transformToFDC3Context(message);
                await this.fdc3.broadcast(fdc3Context);
            }
        } catch (error) {
            throw new Error(`Failed to send message via FDC3: ${error}`);
        }
    }

    subscribeToMessages(channelName: string): Observable<IMessage> {
        if (!this.messageSubjects.has(channelName)) {
            this.messageSubjects.set(channelName, new Subject<IMessage>());
        }
        
        return this.messageSubjects.get(channelName)!.asObservable();
    }

    isConnected(): boolean {
        return this.connected;
    }

    private async setupFDC3Listeners(): Promise<void> {
        if (!this.fdc3) return;

        this.config.channelMappings.forEach(async (mapping) => {
            if (mapping.direction === 'inbound' || mapping.direction === 'bidirectional') {
                const subject = this.messageSubjects.get(mapping.externalChannel) || new Subject<IMessage>();
                this.messageSubjects.set(mapping.externalChannel, subject);

                if (mapping.externalChannel.startsWith('intent:')) {
                    const intentName = mapping.externalChannel.replace('intent:', '');
                    await this.fdc3!.addIntentListener(intentName, (context: FDC3Context) => {
                        const message = this.transformFromFDC3Context(context, mapping.externalChannel);
                        subject.next(message);
                    });
                } else {
                    await this.fdc3!.addContextListener((context: FDC3Context) => {
                        const message = this.transformFromFDC3Context(context, mapping.externalChannel);
                        subject.next(message);
                    });
                }
            }
        });
    }

    private transformToFDC3Context(message: IMessage): FDC3Context {
        return {
            type: message.type || 'fdc3.message',
            id: { messageId: message.id },
            name: message.channelName,
            timestamp: message.timestamp,
            data: message.data,
            isHandled: message.isHandled
        };
    }

    private transformFromFDC3Context(context: FDC3Context, channelName: string): IMessage {
        return {
            channelName,
            data: context.data || context,
            type: context.type,
            timestamp: context.timestamp || Date.now(),
            id: context.id?.messageId || uuid(),
            isHandled: false
        };
    }
}