Add example for adding tasks in bulk

examples/bulk-add-task/consumer.js

@@ -0,0 +1,11 @@
+const { Consumer } = require('orkid');
+
+async function worker(data, metadata) {
+  console.log(`Processing task from Queue: ${metadata.qname}. Task ID: ${metadata.id}. Data:`, data);
+  const result = Math.random();
+  console.log(`Task ${metadata.id} done!`);
+  return result;
+}
+
+const consumer = new Consumer('bulk-add-example-queue', worker, { consumerOptions: { concurrencyPerInstance: 50 } });
+consumer.start();

examples/bulk-add-task/package.json

@@ -0,0 +1,10 @@
+{
+  "name": "orkid-example",
+  "version": "1.0.0",
+  "keywords": [],
+  "private": true,
+  "license": "MIT",
+  "dependencies": {
+    "orkid": "*"
+  }
+}

examples/bulk-add-task/producer.js

@@ -0,0 +1,21 @@
+const { Producer } = require('orkid');
+
+const producer = new Producer('bulk-add-example-queue');
+
+async function addTasks() {
+  // Each task in the array is a object with a mandatory data property
+  // [{ data: 0}, { data: 1}, { data: 2}, ... { data: 999}]
+  //
+  // For de-duplication, you can optionally set dedupKey property too, like:
+  //     [{ data: 0, dedupKey: "unique"}, { data: 1, dedupKey: "unique"}...]
+  const tasks = Array.from({ length: 1000 }).map((_, i) => ({ data: i }));
+
+  await producer.bulkAddTasks(tasks);
+}
+
+addTasks()
+  .then(() => {
+    console.log('Done');
+    process.exit();
+  })
+  .catch(e => console.error(e));

examples/failure-timeout-retry/consumer.js

@@ -2,6 +2,9 @@ const { Consumer } = require('orkid');
 
 const delay = time => new Promise(res => setTimeout(() => res(), time));
 
+// Note: Setting workerFnTimeoutMs in consumerOptions is totally optional.
+// The default behavior is to let workers run for as long as they need.
+
 async function worker(data, metadata) {
   console.log(`Processing task from Queue: ${metadata.qname}. Task ID: ${metadata.id}. Data:`, data);
 
@@ -28,4 +31,5 @@ async function worker(data, metadata) {
 const consumer = new Consumer('retriable-queue', worker, {
   consumerOptions: { concurrencyPerInstance: 1, maxRetry: 1, workerFnTimeoutMs: 3000 }
 });
+
 consumer.start();

src/producer.ts

@@ -12,7 +12,7 @@ export interface ProducerOptions {
 
 export interface Task {
   data: any;
-  dedupKey: string | null;
+  dedupKey?: string | null;
 }
 
 export class Producer {
@@ -78,7 +78,7 @@ export class Producer {
       const pipeline = this._redis!.pipeline();
 
       for (const t of c) {
-        pipeline.enqueue(this.qname, this._DEDUPSET, JSON.stringify(t.data), t.dedupKey, 0);
+        pipeline.enqueue(this.qname, this._DEDUPSET, JSON.stringify(t.data || null), t.dedupKey || null, 0);
       }
 
       const retval = await pipeline.exec();