feat: auto bind event exchanges to endpoints

spec/messageType-spec.ts

@@ -15,4 +15,16 @@ describe("messageType", () => {
 
         expect(messageType.toString()).toBe("urn:message:Contracts:SubmitOrder")
     })
+
+    it("toExchange should return namespace with name without protocol", () => {
+        const messageType = new MessageType("SubmitOrder")
+
+        expect(messageType.toExchange()).toBe("Messages:SubmitOrder")
+    })
+
+    it("toExchange should support custom namespace", () => {
+        const messageType = new MessageType("SubmitOrder", "Contracts")
+
+        expect(messageType.toExchange()).toBe("Contracts:SubmitOrder")
+    })
 })

spec/receiveEndpoint-spec.ts

@@ -0,0 +1,131 @@
+import { defaultReceiveEndpointOptions, ReceiveEndpoint, ReceiveEndpointOptions } from '../dist/receiveEndpoint';
+import { RabbitMqHostAddress } from '../src/RabbitMqEndpointAddress';
+import { Bus } from '../dist/bus';
+import { ConfirmChannel } from 'amqplib';
+import { ChannelContext } from '../dist/channelContext';
+import { MessageType } from '../dist/messageType';
+
+describe("ReceiveEndpoint", () => {
+    let bus: jasmine.SpyObj<Bus>
+    let context: jasmine.SpyObj<ChannelContext>
+    let channel: jasmine.SpyObj<ConfirmChannel>
+
+    let receiveEndpoint: ReceiveEndpoint
+
+    beforeEach(() => {
+        bus = jasmine.createSpyObj('Bus', ['on'], {
+            hostAddress: new RabbitMqHostAddress({ host: '', virtualHost: '' })
+        });
+        const connection = jasmine.createSpyObj('Connection', ['createConfirmChannel']);
+        channel = jasmine.createSpyObj('ConfirmChannel', [
+            'on',
+            'prefetch',
+            'consume',
+            'assertExchange',
+            'assertQueue',
+            'bindQueue',
+            'bindExchange'
+        ])
+        channel.assertQueue.and.returnValue({queue: '', messageCount: 0, consumerCount: 0} as any)
+        channel.consume.and.returnValue({ consumerTag: '' } as any)
+        context = jasmine.createSpyObj('ChannelContext', [], {
+            channel
+        })
+        connection.createConfirmChannel.and.returnValue(channel)
+    })
+
+    it("should configure default prefetch on connect", async () => {
+        createReceiveEndpoint()
+
+        await callOnChannel()
+
+        expect(channel.prefetch).toHaveBeenCalledWith(defaultReceiveEndpointOptions.prefetchCount, defaultReceiveEndpointOptions.globalPrefetch)
+    })
+
+    describe("should configure prefetch on connect", () => {
+        const testCases: [number, boolean][] = [
+            [1, false],
+            [1, true],
+            [5, true]
+        ]
+        for (const [prefetchCount, globalPrefetch] of testCases) {
+            it(`(${prefetchCount}, ${globalPrefetch})`, async () => {
+                const options = {
+                    ...defaultReceiveEndpointOptions,
+                    prefetchCount,
+                    globalPrefetch
+                }
+                createReceiveEndpoint('', options)
+
+                await callOnChannel()
+
+                expect(channel.prefetch).toHaveBeenCalledWith(prefetchCount, globalPrefetch)
+            })
+        }
+    })
+
+    describe("should assertExchange on connect", () => {
+        const testCases: string[] = ["queue name", "another queue"]
+        for (const queueName of testCases) {
+            it(`(${queueName})`, async () => {
+                createReceiveEndpoint(queueName)
+
+                await callOnChannel()
+
+                expect(channel.assertExchange).toHaveBeenCalledWith(queueName, 'fanout', defaultReceiveEndpointOptions)
+            })
+        }
+    })
+
+    describe("should assertQueue on connect", () => {
+        const testCases: string[] = ["queue name", "another queue"]
+        for (const queueName of testCases) {
+            it(`(${queueName})`, async () => {
+                createReceiveEndpoint(queueName)
+
+                await callOnChannel()
+
+                expect(channel.assertQueue).toHaveBeenCalledWith(queueName, defaultReceiveEndpointOptions)
+            })
+        }
+    })
+
+    describe("should bindQueue on connect", () => {
+        const testCases: string[] = ["queue name", "another queue"]
+        for (const queueName of testCases) {
+            it(`(${queueName})`, async () => {
+                createReceiveEndpoint(queueName)
+
+                await callOnChannel()
+
+                expect(channel.bindQueue).toHaveBeenCalledWith(queueName, queueName, '')
+            })
+        }
+    })
+
+    describe("should bind exchanges for every event bound via handle", () => {
+        const testCases: [string, MessageType][] = [
+            ["queue name", new MessageType("Event")],
+            ["another queue", new MessageType("Message", "Contract")]
+        ]
+        for (const [queueName, messageType] of testCases) {
+            it(`(${queueName})`, async () => {
+                createReceiveEndpoint(queueName)
+                receiveEndpoint.handle(messageType, jasmine.createSpy())
+
+                await callOnChannel()
+
+                expect(channel.bindExchange).toHaveBeenCalledWith(queueName, messageType.toExchange(), '')
+            })
+        }
+    })
+
+    async function callOnChannel(): Promise<void> {
+        // TODO: this is a hack so we can actually test the channel setup
+        await (receiveEndpoint as any).onChannel(context);
+    }
+
+    function createReceiveEndpoint(queueName: string = '', options?: ReceiveEndpointOptions) {
+        receiveEndpoint = new ReceiveEndpoint(bus, queueName, undefined, options)
+    }
+})

spec/transport-spec.ts

@@ -0,0 +1,84 @@
+import { Transport } from '../dist/transport';
+import { Bus } from '../dist/bus';
+import { ConnectionContext } from '../dist/connectionContext';
+import { ConfirmChannel } from 'amqplib';
+import { SendContext } from '../dist/sendContext';
+import { RabbitMqHostAddress } from '../src/RabbitMqEndpointAddress';
+
+describe("Transport", () => {
+    let bus: jasmine.SpyObj<Bus>
+    let context: jasmine.SpyObj<ConnectionContext>
+    let channel: jasmine.SpyObj<ConfirmChannel>
+
+    let transport: Transport
+
+    beforeEach(() => {
+        bus = jasmine.createSpyObj('Bus', ['on'], {
+            hostAddress: new RabbitMqHostAddress({ host: '', virtualHost: '' })
+        });
+        const connection = jasmine.createSpyObj('Connection', ['createConfirmChannel']);
+        context = jasmine.createSpyObj('ConnectionContext', [], {
+            connection
+        })
+        channel = jasmine.createSpyObj('ConfirmChannel', ['publish', 'on'])
+        connection.createConfirmChannel.and.returnValue(channel)
+
+        transport = new Transport(bus)
+    })
+
+    describe("send should publish the given message", () => {
+        const testCases = [
+            ['exchange', 'routingKey'],
+            ['MessageExchange', 'RoutingKey']
+        ]
+        for (const [exchange, routingKey] of testCases) {
+            it(`(${exchange}, ${routingKey})`, async () => {
+                callPublishCallback()
+                await callOnConnect()
+
+                await transport.send(exchange, routingKey, new SendContext({}))
+
+                expect(channel.publish).toHaveBeenCalledWith(exchange, routingKey, jasmine.any(Buffer), jasmine.any(Object), jasmine.any(Function))
+            })
+        }
+    })
+
+    it("send should set the contentType", async () => {
+        callPublishCallback()
+        await callOnConnect()
+
+        await transport.send('', '', new SendContext({}))
+
+        expect(channel.publish).toHaveBeenCalledWith(jasmine.any(String), jasmine.any(String), jasmine.any(Buffer), {
+            persistent: true,
+            contentType: "application/vnd.masstransit+json"
+        }, jasmine.any(Function))
+    })
+
+    it("publishing after onConnect should set the contentType", async () => {
+        callPublishCallback()
+        transport.send('', '', new SendContext({}))
+
+        await callOnConnect()
+
+        expect(channel.publish).toHaveBeenCalledWith(jasmine.any(String), jasmine.any(String), jasmine.any(Buffer), {
+            persistent: true,
+            contentType: "application/vnd.masstransit+json"
+        }, jasmine.any(Function))
+    })
+
+    function callPublishCallback() {
+        channel.publish.and.callFake((exchange, routingKey, content, options, callback) => {
+            if (callback != null) {
+                callback(null, {});
+            }
+            return true;
+        })
+    }
+
+    function callOnConnect() {
+        const callback = bus.on.calls.first().args[1];
+
+        return callback(context);
+    }
+})

src/bus.ts

@@ -84,7 +84,7 @@ class MassTransitBus extends EventEmitter implements Bus {
 
             console.log('Bus stopped', this.hostAddress.toString());
         }
-        catch (e) {
+        catch (e: any) {
             console.error('failed to close bus', e.message);
         }
     }
@@ -135,7 +135,7 @@ class MassTransitBus extends EventEmitter implements Bus {
         console.log('Connecting', this.hostAddress.toString());
 
         try {
-            this.connection = connect(this.hostAddress + '?heartbeat=60');
+            this.connection = connect(this.hostAddress + '?heartbeat=60') as unknown as Promise<Connection>;
 
             let connection = await this.connection;
 
@@ -154,7 +154,7 @@ class MassTransitBus extends EventEmitter implements Bus {
 
             this.emit('connect', {hostAddress: this.hostAddress, connection: connection});
         }
-        catch (e) {
+        catch (e: any) {
             console.error('Connect failed', e.message);
 
             this.scheduleReconnect();

src/messageType.ts

@@ -19,5 +19,9 @@ export class MessageType {
     toMessageType(): Array<string> {
         return [this.toString()];
     }
+
+    toExchange(): string {
+        return `${this.ns}:${this.name}`
+    }
 }
 

src/receiveEndpoint.ts

@@ -44,12 +44,14 @@ export class ReceiveEndpoint extends Transport implements ReceiveEndpointConfigu
             let deserializer = new MessageTypeDeserializer<T>(this);
             this._messageTypes[typeName] = deserializer;
             deserializer.on(listener);
+            this.boundEvents.push(messageType);
         }
 
         return this;
     }
 
     private readonly _messageTypes: MessageMap;
+    private readonly boundEvents: MessageType[] = [];
 
     constructor(bus: Bus, queueName: string, cb?: (cfg: ReceiveEndpointConfigurator) => void, options: ReceiveEndpointOptions = defaultReceiveEndpointOptions) {
         super(bus);
@@ -119,6 +121,10 @@ export class ReceiveEndpoint extends Transport implements ReceiveEndpointConfigu
 
         await channel.bindQueue(this.queueName, this.queueName, '');
 
+        for (const messageType of this.boundEvents) {
+            await channel.bindExchange(this.queueName, messageType.toExchange(), '');
+        }
+
         console.log('Queue:', queue.queue, 'MessageCount:', queue.messageCount, 'ConsumerCount:', queue.consumerCount);
     }
 }

src/transport.ts

@@ -116,6 +116,7 @@ export class Transport extends EventEmitter implements Transport {
             let pendingPublish = this.pendingPublishQueue.shift();
             if (!pendingPublish) break;
 
+            // TODO: resolve or reject pending publishes
             let {exchange, message, routingKey} = pendingPublish;
 
             await this.basicPublish(exchange, routingKey, message);
@@ -127,7 +128,7 @@ export class Transport extends EventEmitter implements Transport {
         if (this.channel) {
             let channel = this.channel;
             return new Promise((resolve, reject) => {
-                const result = channel.publish(exchange, routingKey, body, {persistent: true},
+                const result = channel.publish(exchange, routingKey, body, {persistent: true, contentType: "application/vnd.masstransit+json"},
                     err => {
                         if (err) {
                             reject(err);