LSIF: Replace node-resque with bull (#6062)

sourcegraph · Oct 17, 2019 · f022e03 · f022e03
1 parent 5596c5a
commit f022e03
Show file tree

Hide file tree

Showing 9 changed files with 175 additions and 440 deletions.
diff --git a/lsif/package.json b/lsif/package.json
@@ -6,10 +6,10 @@
   "version": "0.1.0",
   "dependencies": {
     "ajv": "^6.10.2",
-    "async-exit-hook": "^2.0.1",
     "async-middleware": "^1.2.1",
     "bloomfilter": "^0.0.18",
     "body-parser": "^1.19.0",
+    "bull": "^3.11.0",
     "express": "^4.17.1",
     "express-opentracing": "^0.1.1",
     "express-winston": "^3.3.0",
@@ -22,7 +22,6 @@
     "logform": "^2.1.2",
     "lsif-protocol": "0.4.3",
     "mz": "^2.7.0",
-    "node-resque": "^5.5.7",
     "on-finished": "^2.3.0",
     "opentracing": "^0.14.4",
     "p-retry": "^4.1.0",
@@ -42,6 +41,7 @@
     "@sourcegraph/tsconfig": "^4.0.0",
     "@types/bloomfilter": "0.0.0",
     "@types/body-parser": "1.17.0",
+    "@types/bull": "^3.10.3",
     "@types/express": "4.17.0",
     "@types/express-winston": "^3.0.4",
     "@types/got": "9.6.7",

diff --git a/lsif/src/@types/async-exit-hook/index.d.ts b/lsif/src/@types/async-exit-hook/index.d.ts
diff --git a/lsif/src/@types/node-resque/index.d.ts b/lsif/src/@types/node-resque/index.d.ts
diff --git a/lsif/src/queue.ts b/lsif/src/queue.ts
@@ -1,27 +1,43 @@
-import { Queue } from 'node-resque'
+import Bull, { Queue, Job } from 'bull'
 import { Span, Tracer, FORMAT_TEXT_MAP } from 'opentracing'
+import { Logger } from 'winston'
 
 /**
- * Enqueue a job to be run by the worker.
+ * Creates a queue instance.
+ *
+ * @param name The name of the queue.
+ * @param endpoint The host:port redis address.
+ * @param logger The logger instance.
+ */
+export function createQueue(name: string, endpoint: string, logger: Logger): Queue {
+    const [host, port] = endpoint.split(':', 2)
+
+    const redis = {
+        host,
+        port: parseInt(port, 10),
+        namespace: `lsif_${name}`,
+    }
+
+    const queue = new Bull(name, { redis })
+    queue.on('error', (error: Error) => logger.error('queue error', { error }))
+    queue.on('global:stalled', (id: string) => logger.error('job stalled', { jobId: id }))
+
+    return queue
+}
+
+/**
+ * Enqueue a job to be run by a worker.
  *
  * @param queue The job queue.
- * @param job The job name.
  * @param args The job arguments.
  * @param tracer The tracer instance.
  * @param span The parent span.
  */
-export const enqueue = (
-    queue: Queue,
-    job: string,
-    args: { [K: string]: any },
-    tracer?: Tracer,
-    span?: Span
-): Promise<void> => {
+export const enqueue = (queue: Queue, args: object, tracer?: Tracer, span?: Span): Promise<Job> => {
+    const tracing = {}
     if (tracer && span) {
-        const tracing = {}
         tracer.inject(span, FORMAT_TEXT_MAP, tracing)
-        args.tracing = tracing
     }
 
-    return queue.enqueue('lsif', job, [args])
+    return queue.add({ args, tracing })
 }