add ignoreOldEvents and maxAge

Author: eli0shin

src/createFunctions.js

@@ -5,6 +5,7 @@ const parseMessage = require('./pubSub/parseMessage');
 const withIdempotency = require('./ensureIdempotent');
 const keepFunctionAlive = require('./keepFunctionAlive');
 const configStore = require('./configStore');
+const ignoreOldEvents = require('./ignoreOldEvents');
 
 const fromEntries = [(acc, [key, value]) => ({ ...acc, [key]: value }), {}];
 
@@ -57,21 +58,24 @@ const setupKeepAlive = (config, service) =>
       }
     : {};
 
-const setupSchedule = (config, service) => ({
-  time,
-  name,
-  function: toExecute,
-  timeZone = 'America/New_York',
-  runtimeOptions,
-}) => [
-  [`${service.basePath}_${name}`],
-  functions
-    .region(config.region)
-    .runWith(runtimeOptions || service.runtimeOptions || {})
-    .pubsub.schedule(time)
-    .timeZone(timeZone)
-    .onRun(toExecute),
-];
+const setupSchedule =
+  (config, service) =>
+  ({
+    time,
+    name,
+    function: toExecute,
+    timeZone = 'America/New_York',
+    runtimeOptions,
+  }) =>
+    [
+      [`${service.basePath}_${name}`],
+      functions
+        .region(config.region)
+        .runWith(runtimeOptions || service.runtimeOptions || {})
+        .pubsub.schedule(time)
+        .timeZone(timeZone)
+        .onRun(toExecute),
+    ];
 
 const setupSchedules = (config, service) =>
   Array.isArray(service.schedule)
@@ -80,30 +84,37 @@ const setupSchedules = (config, service) =>
         .reduce(...fromEntries)
     : {};
 
-const setupEvent = (config, service) => ({
-  topic,
-  type = '',
-  function: toExecute,
-  ensureIdempotent = false,
-  runtimeOptions,
-}) => {
-  const functionName = `${service.basePath}_${topic}${type ? `_${type}` : ''}`;
-
-  return [
-    functionName,
-    functions
-      .region(config.region)
-      .runWith(runtimeOptions || service.runtimeOptions || {})
-      .pubsub.topic(topic)
-      .onPublish(
-        parseMessage(
-          ensureIdempotent
-            ? withIdempotency(functionName, toExecute)
-            : toExecute
-        )
-      ),
-  ];
-};
+const setupEvent =
+  (config, service) =>
+  ({
+    topic,
+    type = '',
+    function: toExecute,
+    ensureIdempotent = false,
+    maxAge,
+    runtimeOptions,
+  }) => {
+    const functionName = `${service.basePath}_${topic}${
+      type ? `_${type}` : ''
+    }`;
+
+    const handlerWithIdempotency = ensureIdempotent
+      ? withIdempotency(functionName, toExecute)
+      : toExecute;
+
+    const handler = Boolean(maxAge)
+      ? ignoreOldEvents(maxAge, handlerWithIdempotency)
+      : handlerWithIdempotency;
+
+    return [
+      functionName,
+      functions
+        .region(config.region)
+        .runWith(runtimeOptions || service.runtimeOptions || {})
+        .pubsub.topic(topic)
+        .onPublish(parseMessage(handler)),
+    ];
+  };
 
 const setupEvents = (config, service) =>
   Array.isArray(service.events)
@@ -122,7 +133,7 @@ const createFunctions = (config = {}, services) => {
   configStore.config = { ...configStore.config, ...config };
 
   return services
-    .map(service => parseConfig(configStore.config, service))
+    .map((service) => parseConfig(configStore.config, service))
     .reduce(...flattenObjects);
 };
 

src/ignoreOldEvents.js

@@ -0,0 +1,24 @@
+/**
+ * If an event continues to retry for a long period of time
+ * there may be a reason to just drop it and stop retrying
+ * This util, when applied, will bail out of a handler if
+ * it is greater than the maxAge set.
+ */
+
+function setupIgnoreOldEvents(maxAge, handler) {
+  return function ignoreOldEvents(...args) {
+    const [_message, context] = args;
+    const eventAgeMs = Date.now() - Date.parse(context.timestamp);
+
+    if (eventAgeMs > maxAge) {
+      console.log(
+        `Dropping event ${context.eventId} with age[ms]: ${eventAgeMs}`
+      );
+      return true;
+    }
+
+    return handler(...args);
+  };
+}
+
+module.exports = setupIgnoreOldEvents;