Move queue into its own plugin

hifiwi-fi · Jul 28, 2022 · 7daf7d2 · 7daf7d2
1 parent a315361
commit 7daf7d2
Show file tree

Hide file tree

Showing 4 changed files with 40 additions and 32 deletions.
diff --git a/lib/queue.js b/lib/queue.js
diff --git a/plugins/queue.js b/plugins/queue.js
@@ -0,0 +1,37 @@
+import fp from 'fastify-plugin'
+import PQueue from 'p-queue'
+
+export default fp(async function (fastify, opts) {
+  const queue = new PQueue({
+    concurrency: 10,
+    timeout: 60000
+  })
+
+  let count = 0
+
+  queue.on('active', () => {
+    fastify.log.info(`Working on item #${++count}.  Size: ${queue.size}  Pending: ${queue.pending}`)
+  })
+
+  queue.on('error', error => {
+    fastify.log.error(error)
+  })
+
+  queue.on('idle', () => {
+    fastify.log.info(`Queue is idle.  Size: ${queue.size}  Pending: ${queue.pending}`)
+  })
+
+  queue.on('add', () => {
+    fastify.log.info(`Task is added.  Size: ${queue.size}  Pending: ${queue.pending}`)
+  })
+
+  queue.on('next', () => {
+    fastify.log.info(`Task is completed.  Size: ${queue.size}  Pending: ${queue.pending}`)
+  })
+
+  fastify.decorate('pqueue', queue)
+},
+{
+  name: 'queue',
+  dependencies: ['env']
+})
diff --git a/routes/api/bookmarks/index.js b/routes/api/bookmarks/index.js
@@ -1,7 +1,6 @@
 /* eslint-disable camelcase */
 import SQL from '@nearform/sql'
 import { createEpisode } from '../../../lib/create-episode.js'
-import { queue } from '../../../lib/queue.js'
 import { runYTDLP } from '../../../lib/run-yt-dlp.js'
 
 const commnonBookmarkProps = {
@@ -413,7 +412,7 @@ export default async function bookmarkRoutes (fastify, opts) {
 
           await client.query('commit')
 
-          queue.add(runYTDLP({
+          fastify.pqueue.add(runYTDLP({
             userId,
             bookmarkId: bookmark.id,
             episodeId,
@@ -586,7 +585,7 @@ export default async function bookmarkRoutes (fastify, opts) {
 
         await client.query('commit')
 
-        queue.add(runYTDLP({
+        fastify.pqueue.add(runYTDLP({
           userId,
           bookmarkId,
           episodeId,

diff --git a/routes/api/feeds/index.js b/routes/api/feeds/index.js
@@ -320,7 +320,7 @@ export default async function podcastFeedsRoutes (fastify, opts) {
         reply.header('fly-cache-status', 'MISS')
       }
 
-      const metadata = await getYTDLPUrl({ url: episode.src_url })
+      const metadata = await fastify.pqueue.add(() => getYTDLPUrl({ url: episode.src_url }))
       cache.set(cacheKey, metadata.urls, metadata.urls)
       reply.redirect(302, metadata.urls)
     }