Fix: Memory queue fail deletes other messages (#163)

* fix memory-queue fail handling

* package all from service-bus

Author: adenhertog

packages/bus-core/src/index.ts

@@ -1,5 +1,3 @@
-export * from './service-bus/bus'
-export * from './service-bus/bus-instance'
 export {
   Transport,
   TransportMessage,
@@ -13,3 +11,4 @@ export * from './workflow'
 export * from './error'
 export * from './container'
 export * from './logger'
+export * from './service-bus'

packages/bus-core/src/transport/memory-queue.spec.ts

@@ -5,8 +5,10 @@ import { MessageAttributes } from '@node-ts/bus-messages'
 import * as faker from 'faker'
 import { IMock, Mock } from 'typemoq'
 import { Logger, LoggerFactory } from '../logger'
-import { DefaultHandlerRegistry, HandlerRegistry } from '../handler'
+import { DefaultHandlerRegistry, handlerFor, HandlerRegistry } from '../handler'
 import { JsonSerializer, MessageSerializer } from '../serialization'
+import EventEmitter from 'events'
+import { Bus } from '../service-bus/bus'
 
 const event = new TestEvent()
 const command = new TestCommand()
@@ -163,5 +165,24 @@ describe('MemoryQueue', () => {
     it('should forward it to the dead letter queue', () => {
       expect(sut.deadLetterQueueDepth).toEqual(1)
     })
+
+    it('should only fail the handled message', async () => {
+      const emitter = new EventEmitter()
+      const bus = await Bus.configure()
+        .withConcurrency(1)
+        .withHandler(handlerFor(TestEvent, async () => {
+          await bus.send(new TestCommand())
+          await bus.fail()
+        }))
+        .withHandler(handlerFor(TestCommand, () => { emitter.emit('done') }))
+        .initialize()
+
+      await bus.start()
+
+      const completion = new Promise<void>(resolve => emitter.once('done', resolve))
+      await bus.publish(new TestEvent())
+      await completion
+      await bus.dispose()
+    })
   })
 })

packages/bus-core/src/transport/memory-queue.ts

@@ -40,7 +40,7 @@ export class MemoryQueue implements Transport<InMemoryMessage> {
 
   private queue: TransportMessage<InMemoryMessage>[] = []
   private queuePushed: EventEmitter = new EventEmitter()
-  private deadLetterQueue: TransportMessage<InMemoryMessage>[] = []
+  private _deadLetterQueue: TransportMessage<InMemoryMessage>[] = []
   private messagesWithHandlers: { [key: string]: {} }
   private logger: Logger
   private coreDependencies: CoreDependencies
@@ -116,6 +116,11 @@ export class MemoryQueue implements Transport<InMemoryMessage> {
 
   async deleteMessage (message: TransportMessage<InMemoryMessage>): Promise<void> {
     const messageIndex = this.queue.indexOf(message)
+    if (messageIndex < 0) {
+      // actions like .fail() will cause the message to already be deleted
+      this.logger.debug('Message already deleted', { message, messageIndex })
+      return
+    }
     this.logger.debug('Deleting message', { queueDepth: this.depth, messageIndex })
     this.queue.splice(messageIndex, 1)
     this.logger.debug('Message Deleted', { queueDepth: this.depth })
@@ -141,7 +146,15 @@ export class MemoryQueue implements Transport<InMemoryMessage> {
   }
 
   get deadLetterQueueDepth (): number {
-    return this.deadLetterQueue.length
+    return this._deadLetterQueue.length
+  }
+
+  /**
+   * Returns all messages sitting in the dead letter queue. This is a copy of the queue
+   * so mutative actions on this array will have no consequence.
+   */
+  get deadLetterQueue (): TransportMessage<InMemoryMessage>[] {
+    return [...this._deadLetterQueue]
   }
 
   /**
@@ -153,7 +166,7 @@ export class MemoryQueue implements Transport<InMemoryMessage> {
 
 
   private async sendToDeadLetterQueue (message: TransportMessage<InMemoryMessage>): Promise<void> {
-    this.deadLetterQueue.push(message)
+    this._deadLetterQueue.push(message)
     await this.deleteMessage(message)
   }