Version 2 (published as next to NPM)

.github/workflows/build.yaml

@@ -0,0 +1,91 @@
+name: Build
+
+on: [push]
+
+env:
+  NODE_OPTIONS: --max_old_space_size=4096
+
+jobs:
+
+  test:
+    name: Test
+    strategy:
+      matrix:
+        version: [16, 18, 20]
+        os: [ubuntu-latest]
+    runs-on: ${{ matrix.os }}
+    services:
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
+        options: --health-cmd="redis-cli ping" --health-interval=10s --health-timeout=5s --health-retries=5
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-node@v3
+      with:
+        node-version: ${{ matrix.version }}
+    - uses: pnpm/action-setup@v2
+      name: Install pnpm
+      with:
+        version: 8
+        run_install: false
+    - name: Get pnpm store directory
+      shell: bash
+      run: |
+        echo "STORE_PATH=$(pnpm store path --silent)" >> $GITHUB_ENV
+    - uses: actions/cache@v3
+      name: Setup pnpm cache
+      with:
+        path: ${{ env.STORE_PATH }}
+        key: ${{ runner.os }}-pnpm-store-${{ hashFiles('**/pnpm-lock.yaml') }}
+        restore-keys: |
+          ${{ runner.os }}-pnpm-store-
+    - run: pnpm install
+    - run: npm test
+
+  build:
+    name: Coverage and Sentry
+    needs: [test]
+    runs-on: ubuntu-latest
+    services:
+      redis:
+        image: redis
+        ports:
+          - 6379:6379
+        options: --health-cmd="redis-cli ping" --health-interval=10s --health-timeout=5s --health-retries=5
+    steps:
+    - uses: actions/checkout@v3
+    - uses: actions/setup-node@v3
+      with:
+        node-version: ${{ matrix.version }}
+    - uses: pnpm/action-setup@v2
+      name: Install pnpm
+      with:
+        version: 8
+        run_install: false
+    - name: Get pnpm store directory
+      shell: bash
+      run: |
+        echo "STORE_PATH=$(pnpm store path --silent)" >> $GITHUB_ENV
+    - uses: actions/cache@v3
+      name: Setup pnpm cache
+      with:
+        path: ${{ env.STORE_PATH }}
+        key: ${{ runner.os }}-pnpm-store-${{ hashFiles('**/pnpm-lock.yaml') }}
+        restore-keys: |
+          ${{ runner.os }}-pnpm-store-
+    - run: pnpm install
+    - run: npm test -- --coverage
+    - name: Upload coverage reports to Codecov
+      uses: codecov/codecov-action@v3
+      env:
+        CODECOV_TOKEN: ${{ secrets.CODECOV_TOKEN }}
+    - name: Create Sentry release
+      uses: getsentry/[email protected]
+      env:
+        SENTRY_AUTH_TOKEN: ${{ secrets.SENTRY_AUTH_TOKEN }}
+        SENTRY_ORG: suredone
+        SENTRY_PROJECT: qdone
+      with:
+        version: ${{ github.sha }}

.gitignore

@@ -1,2 +1,7 @@
 node_modules/
 .nyc_output/
+coverage
+.DS_Store
+*.swp
+commonjs/src
+commonjs/*.js

README.md

@@ -429,6 +429,10 @@ Options
                               option if you are managing retries outside of qdone, and make sure the ID is
                               the same for each retry in the deduplication window. Defaults to a string
                               unique to this invocation.
+    -d, --delay number        Delays delivery of each message by the given number of seconds (up to 900 seconds,
+                              or 15 minutes). Defaults to immediate delivery (no delay).
+    --dlq-after number        Sends messages from the failed queue to a dead letter queue after this many failures. [default: 3]
+    --dlq-suffix string       Suffix to append to each queue to generate DLQ name [default: _dead]
     --prefix string           Prefix to place at the front of each SQS queue name [default: qdone_]
     --fail-suffix string      Suffix to append to each queue to generate fail queue name [default: _failed]
     --region string           AWS region for Queues [default: us-east-1]
@@ -482,4 +486,4 @@ If a queue name ends with the * (wildcard) character, worker will listen on all
     -q, --quiet             Turn on production logging. Automatically set if stderr is not a tty.
     -v, --verbose           Turn on verbose output. Automatically set if stderr is a tty.
     -V, --version           Show version number
-    --help                  Print full help message.
+    --help                  Print full help message.

commonjs/package.json

@@ -0,0 +1,3 @@
+{
+  "type": "commonjs"
+}

examples/processMessage.js

@@ -0,0 +1,16 @@
+#!/usr/bin/env -S node --experimental-json-modules
+//
+// This example enqueues three messages and processes them.
+//
+import { enqueue, processMessages, requestShutdown } from '../index.mjs' // from 'qdone' for standalone example
+
+await enqueue('test1', JSON.stringify({ one: 1 }))
+await enqueue('test2', JSON.stringify({ two: 2 }))
+await enqueue('test3', JSON.stringify({ three: 3 }))
+
+async function callback (queue, payload) {
+  console.log({ queue, payload })
+  if (payload.three) requestShutdown()
+}
+
+await processMessages(['test1', 'test2', 'test3'], callback)

examples/processMessageRandom.js

@@ -0,0 +1,19 @@
+#!/usr/bin/env -S node --experimental-json-modules
+//
+// This example processes random messages once a second until you Ctl-C
+//
+
+import { enqueue, processMessages, requestShutdown } from '../index.mjs' // from 'qdone' for standalone example
+
+const randomEnqueue = setInterval(function () {
+  enqueue(['rtest1', 'rtest2', 'rtest3'][Math.round(Math.random() * 2)], JSON.stringify({ foo: Math.round(Math.random() * 10) }))
+}, 1000)
+
+process.on('SIGINT', () => { clearInterval(randomEnqueue); console.log('SIGINT'); requestShutdown() })
+process.on('SIGTERM', () => { clearInterval(randomEnqueue); console.log('SIGTERM'); requestShutdown() })
+
+async function callback (queue, payload) {
+  console.log({ queue, payload })
+}
+
+await processMessages(['rtest1', 'rtest2', 'rtest3'], callback, { verbose: true, disableLog: true })

examples/simpleScheduler.js

@@ -0,0 +1,49 @@
+#!/usr/bin/env -S node --experimental-json-modules
+
+//
+// This example implements a basic scheduler on top of qdone. The
+// processMessages callback looks at the queue name throws a DoNotProcess
+// error if there are already two messages processing on this queue.
+//
+
+import {
+  enqueue,
+  processMessages,
+  requestShutdown,
+  DoNotProcess
+} from '../index.mjs' // from 'qdone' for standalone example
+
+const randomEnqueue = setInterval(function () {
+  const queue = ['rtest1', 'rtest2', 'rtest3'][Math.round(Math.random() * 2)]
+  const payload = { foo: Math.round(Math.random() * 10) }
+  console.log({ enqueue: { queue, payload } })
+  enqueue(queue, JSON.stringify(payload))
+}, 200)
+
+process.on('SIGINT', () => { clearInterval(randomEnqueue); console.log('SIGINT'); requestShutdown() })
+process.on('SIGTERM', () => { clearInterval(randomEnqueue); console.log('SIGTERM'); requestShutdown() })
+
+// This returns a promise that resolves in the given number of milliseconds
+const delay = (ms) => new Promise(resolve => setTimeout(resolve, ms))
+
+// This keeps track of the number of active jobs per queue
+const activeCount = {}
+
+async function callback (queue, payload) {
+  const numActive = activeCount[queue] = (activeCount[queue] || 0) + 1
+  console.log({ numActive, activeCount })
+  try {
+    // Limit to 2 active tasks per queue
+    if (numActive > 2) {
+      console.log({ refusing: { queue, payload } })
+      throw new DoNotProcess()
+    }
+
+    console.log({ processing: { queue, payload } })
+    await delay(Math.random() * 1000 * 2) // Processing takes up to 2 seconds
+  } finally {
+    activeCount[queue] = (activeCount[queue] || 0) - 1
+  }
+}
+
+await processMessages(['rtest1', 'rtest2', 'rtest3'], callback, { verbose: true, disableLog: true })

index.js

@@ -1,6 +1,2 @@
-
-module.exports = {
-  enqueue: require('./src/enqueue'),
-  worker: require('./src/worker'),
-  cli: require('./src/cli')
-}
+export { enqueue, enqueueBatch } from './src/enqueue.js'
+export { processMessages, requestShutdown, DoNotProcess } from './src/consumer.js'