feat: added client input

.husky/pre-commit

@@ -1,4 +1,4 @@
 #!/bin/sh
 . "$(dirname "$0")/_/husky.sh"
 
-pnpm format:check && pnpm lint:check && pnpm test:coverage
+pnpm format:check && pnpm lint:check && pnpm test

package.json

@@ -18,16 +18,18 @@
     "access": "public"
   },
   "scripts": {
-    "lint:check": "eslint src/** --ext=ts",
-    "lint:fix": "eslint src/** --fix --ext=ts",
+    "build": "parcel build",
     "format:check": "prettier --check src/**",
     "format:fix": "prettier --write src/**",
+    "lint:check": "eslint src/** --ext=ts",
+    "lint:fix": "eslint src/** --fix --ext=ts",
     "prepare": "husky install",
-    "watch": "parcel watch",
-    "build": "parcel build",
-    "test:dev": "vitest",
+    "prepublish": "pnpm build",
     "test:coverage": "vitest run --coverage",
-    "prepublish": "pnpm build"
+    "test:dev": "vitest",
+    "test": "pnpm test:coverage",
+    "typecheck": "vitest typecheck --run",
+    "watch": "parcel watch"
   },
   "keywords": [
     "azure",

src/index.ts

@@ -2,22 +2,21 @@ import { DequeuedMessageItem, QueueClient, QueueServiceClient } from '@azure/sto
 import { HandlerFunction, QueueError, QueueOptions } from './utils'
 import { QueueEventEmitter } from './events'
 
+type QueueConnection = string | QueueServiceClient
+
 export class AzureQueueConsumer extends QueueEventEmitter {
   #options: QueueOptions
   #handler: HandlerFunction
   #queueClient: QueueClient
   #pollingTime: number
   #shouldShutdown = false
-  constructor(queueName: string, connectionString: string, handler: HandlerFunction, options?: QueueOptions) {
+  constructor(queueName: string, connection: QueueConnection, handler: HandlerFunction, options?: QueueOptions) {
     super()
     this.#handler = handler
     this.#options = options ?? { pollingTime: 10 }
     this.#options.maxTries = this.#options.maxTries ?? 4
     this.#options.numberOfMessages = this.#options.numberOfMessages ?? 1
-    const queueServiceClient = QueueServiceClient.fromConnectionString(connectionString, {
-      retryOptions: { maxTries: this.#options.maxTries }
-    })
-    this.#queueClient = queueServiceClient.getQueueClient(queueName)
+    this.#queueClient = this.#getQueueClient(connection, queueName)
     this.#pollingTime = this.#options.pollingTime
     this.#createQueueAsync()
   }
@@ -77,4 +76,16 @@ export class AzureQueueConsumer extends QueueEventEmitter {
     this.#shouldShutdown = true
     this.emit('queue::shutdown')
   }
+
+  #getQueueClient = (connection: QueueConnection, queueName: string) => {
+    let queueServiceClient: QueueServiceClient
+    if (typeof connection == 'string') {
+      queueServiceClient = QueueServiceClient.fromConnectionString(connection, {
+        retryOptions: { maxTries: this.#options.maxTries }
+      })
+    } else {
+      queueServiceClient = connection
+    }
+    return queueServiceClient.getQueueClient(queueName)
+  }
 }

tests/consumer.test.ts

@@ -1,15 +1,16 @@
 import { afterAll, afterEach, describe, expect, it, vi } from 'vitest'
-import { QUEUE_MESSAGE } from './fixtures/Message'
 import { AzureQueueConsumer } from '../src'
-import { DequeuedMessageItem, QueueServiceClient } from '@azure/storage-queue'
-import { flushPromises } from './helper/helper'
+import { QueueServiceClient } from '@azure/storage-queue'
+import { flushPromises, QueueService } from './helper/helper'
+
+const queueService = new QueueService()
 
 vi.mock('@azure/storage-queue', () => {
   const QueueServiceClient = vi.fn()
-  QueueServiceClient.fromConnectionString = vi.fn((...args) => {
+  QueueServiceClient['fromConnectionString'] = vi.fn((...args) => {
     const getQueueClient = vi.fn(() => {
       const createIfNotExists = vi.fn(() => new Promise((resolve) => resolve('this')))
-      const receiveMessages = vi.fn(() => new Promise((resolve) => resolve(QUEUE_MESSAGE)))
+      const receiveMessages = vi.fn(() => new Promise((resolve) => resolve(queueService.fetchMessages(2))))
       const deleteMessage = vi.fn(
         (messageId: string, popReceipt: string) => new Promise((resolve) => resolve(messageId))
       )
@@ -44,7 +45,7 @@ describe('azure queue tests', () => {
       const errorToThrow = { code: 'code', message: 'message' }
 
       try {
-        vi.spyOn(QueueServiceClient, 'fromConnectionString').mockImplementation((...args) => {
+        vi.spyOn(QueueServiceClient, 'fromConnectionString').mockImplementation((...args: string[]) => {
           const getQueueClient = vi.fn(() => {
             const createIfNotExists = vi.fn(() => new Promise((resolve, reject) => reject(errorToThrow)))
 
@@ -68,7 +69,7 @@ describe('azure queue tests', () => {
       listener.listen()
       await flushPromises()
       expect(handler).toHaveBeenCalledTimes(1)
-      expect(handler).toHaveBeenCalledWith(QUEUE_MESSAGE.receivedMessageItems)
+      expect(handler).toHaveBeenCalledWith(queueService.fetchMessages(2).receivedMessageItems)
     })
     describe('listener event tests', () => {
       it('should emit event when message is received', async () => {
@@ -80,7 +81,7 @@ describe('azure queue tests', () => {
 
         await flushPromises()
         expect(handler).toBeCalledTimes(1)
-        expect(handler).toHaveBeenCalledWith(QUEUE_MESSAGE)
+        expect(handler).toHaveBeenCalledWith(queueService.fetchMessages(2))
       })
       it('should emit event when handler completes execution', async () => {
         expect.assertions(2)
@@ -120,13 +121,13 @@ describe('azure queue tests', () => {
 
         expect(handler).toHaveBeenNthCalledWith(
           1,
-          QUEUE_MESSAGE.receivedMessageItems[0].messageId,
-          QUEUE_MESSAGE.receivedMessageItems[0].popReceipt
+          queueService.fetchMessages(2).receivedMessageItems[0].messageId,
+          queueService.fetchMessages(2).receivedMessageItems[0].popReceipt
         )
         expect(handler).toHaveBeenNthCalledWith(
           2,
-          QUEUE_MESSAGE.receivedMessageItems[1].messageId,
-          QUEUE_MESSAGE.receivedMessageItems[1].popReceipt
+          queueService.fetchMessages(2).receivedMessageItems[1].messageId,
+          queueService.fetchMessages(2).receivedMessageItems[1].popReceipt
         )
         expect(handler).toHaveBeenCalledTimes(2)
       })
@@ -137,8 +138,14 @@ describe('azure queue tests', () => {
         listener.on('message::afterDelete', handler)
         listener.listen()
         await flushPromises()
-        expect(handler).toHaveBeenNthCalledWith(1, QUEUE_MESSAGE.receivedMessageItems[0].messageId)
-        expect(handler).toHaveBeenNthCalledWith(2, QUEUE_MESSAGE.receivedMessageItems[1].messageId)
+        expect(handler).toHaveBeenNthCalledWith(
+          1,
+          queueService.fetchMessages(2).receivedMessageItems[0].messageId
+        )
+        expect(handler).toHaveBeenNthCalledWith(
+          2,
+          queueService.fetchMessages(2).receivedMessageItems[1].messageId
+        )
         expect(handler).toHaveBeenCalledTimes(2)
       })
     })

tests/helper/helper.ts

@@ -1 +1,46 @@
+import { DequeuedMessageItem, QueueCreateIfNotExistsResponse, QueueReceiveMessageResponse } from '@azure/storage-queue'
+
+export class QueueService {
+  readonly messages: DequeuedMessageItem[] = [
+    {
+      messageId: 'abcderhbub',
+      insertedOn: new Date(),
+      expiresOn: new Date(),
+      popReceipt: '#6788888',
+      nextVisibleOn: new Date(),
+      dequeueCount: 0,
+      messageText: '{"hello":"world","a":5}'
+    },
+    {
+      messageId: 'gjhcvf',
+      insertedOn: new Date(),
+      expiresOn: new Date(),
+      popReceipt: '#1234',
+      nextVisibleOn: new Date(),
+      dequeueCount: 0,
+      messageText: '{"world":"hello","a":5}'
+    }
+  ]
+  queueMessage: DequeuedMessageItem[] = []
+  constructor() {
+    this.queueMessage = this.messages
+  }
+
+  deleteMessage = (messageId: string) => {
+    this.queueMessage = this.queueMessage.filter((message) => message.messageId != messageId)
+  }
+
+  restoreMessages = () => {
+    this.queueMessage = this.messages
+  }
+
+  fetchMessages = (count: number): QueueReceiveMessageResponse => ({
+    receivedMessageItems: this.queueMessage.slice(0, Math.min(count, this.queueMessage.length)),
+    _response: undefined,
+    errorCode: undefined
+  })
+
+  createQueue = (): QueueCreateIfNotExistsResponse => ({ succeeded: true, _response: undefined })
+}
+
 export const flushPromises = async () => await new Promise(process.nextTick)

vite.config.ts

@@ -12,4 +12,4 @@ export default defineConfig({
       exclude: ['dist/**', 'tests/**']
     }
   }
-})
+})