Implement cache notifier (#6)

Author: codeaholicguy

index.js

@@ -2,11 +2,10 @@
 
 const fp = require('fastify-plugin')
 const Keyv = require('keyv')
-const BPromise = require('bluebird')
 const crypto = require('crypto')
+const {EventEmitter} = require('events')
 
 const CACHEABLE_METHODS = ['GET']
-const INTERVAL = 200
 const X_RESPONSE_CACHE = 'x-response-cache'
 const X_RESPONSE_CACHE_HIT = 'hit'
 const X_RESPONSE_CACHE_MISS = 'miss'
@@ -27,36 +26,41 @@ function buildCacheKey(req, {headers}) {
   return key
 }
 
-async function waitForCacheFulfilled(cache, key, timeout) {
-  let cachedString = await cache.get(key)
-  let waitedFor = 0
-
-  while (!cachedString) {
-    await BPromise.delay(INTERVAL)
-    cachedString = await cache.get(key)
-
-    waitedFor += INTERVAL
-    if (!cachedString && waitedFor > timeout) {
-      return
-    }
-  }
-
-  return cachedString
-}
-
 function createOnRequestHandler({ttl, additionalCondition: {headers}}) {
   return async function handler(req, res) {
     if (!isCacheableRequest(req)) {
       return
     }
 
     const cache = this.responseCache
+    const cacheNotifier = this.responseCacheNotifier
     const key = buildCacheKey(req, {headers})
     const requestKey = `${key}__requested`
     const isRequestExisted = await cache.get(requestKey)
 
+    async function waitForCacheFulfilled(key) {
+      return new Promise((resolve) => {
+        cache.get(key).then((cachedString) => {
+          if (cachedString) {
+            resolve(cachedString)
+          }
+        })
+
+        const handler = async () => {
+          const cachedString = await cache.get(key)
+
+          resolve(cachedString)
+        }
+
+        cacheNotifier.once(key, handler)
+
+        setTimeout(() => cacheNotifier.removeListener(key, handler), ttl)
+        setTimeout(() => resolve(), ttl)
+      })
+    }
+
     if (isRequestExisted) {
-      const cachedString = await waitForCacheFulfilled(cache, key, ttl)
+      const cachedString = await waitForCacheFulfilled(key)
 
       if (cachedString) {
         const cached = JSON.parse(cachedString)
@@ -80,6 +84,7 @@ function createOnSendHandler({ttl, additionalCondition: {headers}}) {
     }
 
     const cache = this.responseCache
+    const cacheNotifier = this.responseCacheNotifier
     const key = buildCacheKey(req, {headers})
 
     await cache.set(
@@ -90,6 +95,7 @@ function createOnSendHandler({ttl, additionalCondition: {headers}}) {
       }),
       ttl,
     )
+    cacheNotifier.emit(key)
   }
 }
 
@@ -101,8 +107,10 @@ const responseCachingPlugin = (
   const headers = additionalCondition.headers || []
   const opts = {ttl, additionalCondition: {headers}}
   const responseCache = new Keyv()
+  const responseCacheNotifier = new EventEmitter()
 
   instance.decorate('responseCache', responseCache)
+  instance.decorate('responseCacheNotifier', responseCacheNotifier)
   instance.addHook('onRequest', createOnRequestHandler(opts))
   instance.addHook('onSend', createOnSendHandler(opts))
 

index.test.js

@@ -4,10 +4,13 @@ const test = require('tap').test
 
 const axios = require('axios')
 const fastify = require('fastify')
-const BPromise = require('bluebird')
 
 const plugin = require('./index.js')
 
+function delay(ms) {
+  return new Promise((resolve) => setTimeout(resolve, ms))
+}
+
 test('should decorate cache to fastify instance', (t) => {
   t.plan(3)
   const instance = fastify()
@@ -30,7 +33,7 @@ test('should cache the cacheable request', (t) => {
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/cache`
-    const [response1, response2] = await BPromise.all([
+    const [response1, response2] = await Promise.all([
       axios.get(address),
       axios.get(address),
     ])
@@ -55,7 +58,7 @@ test('should not cache the uncacheable request', (t) => {
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/no-cache`
-    const [response1, response2] = await BPromise.all([
+    const [response1, response2] = await Promise.all([
       axios.post(address, {}),
       axios.post(address, {}),
     ])
@@ -80,11 +83,11 @@ test('should apply ttl config', (t) => {
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/ttl`
-    const [response1, response2] = await BPromise.all([
+    const [response1, response2] = await Promise.all([
       axios.get(address),
       axios.get(address),
     ])
-    await BPromise.delay(3000)
+    await delay(3000)
     const response3 = await axios.get(address)
     t.is(response1.status, 200)
     t.is(response2.status, 200)
@@ -114,7 +117,7 @@ test('should apply additionalCondition config', (t) => {
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/headers`
-    const [response1, response2, response3, response4] = await BPromise.all([
+    const [response1, response2, response3, response4] = await Promise.all([
       axios.get(address, {
         headers: {'x-should-applied': 'yes'},
       }),
@@ -146,15 +149,15 @@ test('should waiting for cache if multiple same request come in', (t) => {
   const instance = fastify()
   instance.register(plugin, {ttl: 5000})
   instance.get('/waiting', async (req, res) => {
-    await BPromise.delay(3000)
+    await delay(3000)
     res.send({hello: 'world'})
   })
   instance.listen(0, async (err) => {
     if (err) t.threw(err)
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/waiting`
-    const [response1, response2] = await BPromise.all([
+    const [response1, response2] = await Promise.all([
       axios.get(address),
       axios.get(address),
     ])
@@ -172,15 +175,15 @@ test('should not waiting for cache due to timeout', (t) => {
   const instance = fastify()
   instance.register(plugin)
   instance.get('/abort', async (req, res) => {
-    await BPromise.delay(2000)
+    await delay(2000)
     res.send({hello: 'world'})
   })
   instance.listen(0, async (err) => {
     if (err) t.threw(err)
     instance.server.unref()
     const portNum = instance.server.address().port
     const address = `http://127.0.0.1:${portNum}/abort`
-    const [response1, response2] = await BPromise.all([
+    const [response1, response2] = await Promise.all([
       axios.get(address),
       axios.get(address),
     ])

package-lock.json

@@ -9,7 +9,9 @@
     }
   },
   "lint-staged": {
-    "**/*.js": ["eslint"]
+    "**/*.js": [
+      "eslint"
+    ]
   },
   "scripts": {
     "test": "tap --cov *.test.js",
@@ -33,7 +35,6 @@
   },
   "homepage": "https://github.com/codeaholicguy/fastify-response-caching#readme",
   "dependencies": {
-    "bluebird": "^3.7.2",
     "fastify-plugin": "^2.3.4",
     "keyv": "^4.0.2"
   },