feat: migrate s3 plugin to serverless v4

Author: agnusha

packages/serverless-offline-s3/src/index.js

@@ -1,7 +1,5 @@
 const {get, isUndefined, omitBy, pick} = require('lodash/fp');
 
-const log = require('@serverless/utils/log').log;
-
 const S3 = require('./s3');
 
 const OFFLINE_OPTION = 'serverless-offline';
@@ -17,15 +15,21 @@ const defaultOptions = {
 const omitUndefined = omitBy(isUndefined);
 
 class ServerlessOfflineS3 {
-  constructor(serverless, cliOptions) {
+  constructor(serverless, cliOptions, {log} = {}) {
     this.cliOptions = null;
     this.options = null;
     this.s3 = null;
     this.lambda = null;
     this.serverless = null;
+    this.log = null;
 
     this.cliOptions = cliOptions;
     this.serverless = serverless;
+    this.log = log || {
+      debug: console.debug.bind(console),
+      notice: console.log.bind(console),
+      warning: console.warn.bind(console)
+    };
 
     this.hooks = {
       'offline:start:init': this.start.bind(this),
@@ -52,7 +56,7 @@ class ServerlessOfflineS3 {
 
     await Promise.all(eventModules);
 
-    this.serverless.cli.log(
+    this.log.notice(
       `Starting Offline S3 at stage ${this.options.stage} (${this.options.endPoint}/${this.options.region})`
     );
   }
@@ -68,7 +72,7 @@ class ServerlessOfflineS3 {
 
     signals.map(signal =>
       process.on(signal, async () => {
-        this.serverless.cli.log(`Got ${signal} signal. Offline Halting...`);
+        this.log.notice(`Got ${signal} signal. Offline Halting...`);
 
         await this.end();
       })
@@ -85,7 +89,7 @@ class ServerlessOfflineS3 {
       return;
     }
 
-    this.serverless.cli.log('Halting offline server');
+    this.log.notice('Halting offline server');
 
     const eventModules = [];
 
@@ -114,7 +118,7 @@ class ServerlessOfflineS3 {
   async _createS3(events, skipStart) {
     const resources = this._getResources();
 
-    this.s3 = new S3(this.lambda, resources, this.options);
+    this.s3 = new S3(this.lambda, resources, this.options, this.log);
 
     await this.s3.create(events);
 
@@ -140,7 +144,7 @@ class ServerlessOfflineS3 {
       omitUndefined(this.cliOptions)
     );
 
-    log.debug('options:', this.options);
+    this.log.debug('s3 options:', this.options);
   }
 
   _getEvents() {

packages/serverless-offline-s3/src/s3.js

@@ -1,8 +1,6 @@
 const Minio = require('minio');
 const {assign, toNumber} = require('lodash/fp');
 
-const log = require('@serverless/utils/log').log;
-
 const S3EventDefinition = require('./s3-event-definition');
 const S3Event = require('./s3-event');
 
@@ -11,15 +9,23 @@ const delay = timeout =>
     setTimeout(resolve, timeout);
   });
 
+const defaultLog = {
+  debug: console.debug.bind(console),
+  notice: console.log.bind(console),
+  warning: console.warn.bind(console)
+};
+
 class S3 {
-  constructor(lambda, resources, options) {
+  constructor(lambda, resources, options, log = defaultLog) {
     this.lambda = null;
     this.resources = null;
     this.options = null;
+    this.log = null;
 
     this.lambda = lambda;
     this.resources = resources;
     this.options = options;
+    this.log = log || defaultLog;
 
     const s3Endpoint = this.options.endpoint ? new URL(this.options.endpoint) : {};
     this.client = new Minio.Client(
@@ -34,6 +40,14 @@ class S3 {
     this.listeners = [];
   }
 
+  _safeLog(level, message) {
+    if (this.log && typeof this.log[level] === 'function') {
+      this.log[level](message);
+    } else if (console[level]) {
+      console[level](message);
+    }
+  }
+
   create(events) {
     this.events = events;
     return Promise.all(
@@ -48,35 +62,45 @@ class S3 {
     return Promise.all(
       this.events.map(async ({functionKey, s3}) => {
         const {event, bucket, rules} = s3;
-        await this._waitFor(bucket);
+        this.log.debug(`Setting up listener for bucket: ${bucket}, event: ${event}`);
 
+        await this._waitFor(bucket);
         const eventRules = rules || [];
         const prefix = (eventRules.find(rule => rule.prefix) || {prefix: '*'}).prefix;
         const suffix = (eventRules.find(rule => rule.suffix) || {suffix: '*'}).suffix;
-
         const listener = this.client.listenBucketNotification(bucket, prefix, suffix, [event]);
 
         listener.on('notification', async record => {
           if (record) {
             try {
+              this.log.debug(
+                `Received S3 notification for bucket ${bucket}: ${JSON.stringify(record)}`
+              );
               const lambdaFunction = this.lambda.get(functionKey);
 
               const s3Notification = new S3Event(record);
               lambdaFunction.setEvent(s3Notification);
 
               await lambdaFunction.runHandler();
             } catch (err) {
-              log.warn(err.stack);
+              this.log.warning(
+                `Error processing S3 notification for bucket ${bucket}: ${err.stack}`
+              );
             }
           }
         });
 
+        listener.on('error', err => {
+          this.log.warning(`Error in S3 listener for bucket ${bucket}: ${err.message}`);
+        });
+
         this.listeners = [...this.listeners, listener];
+        this.log.debug(`Listener set up successfully for bucket: ${bucket}`);
       })
     );
   }
 
-  stop(timeout) {
+  stop(_timeout) {
     this.listeners.forEach(listener => listener.stop());
     this.listeners = [];
   }
@@ -114,7 +138,7 @@ class S3 {
 
           await lambdaFunction.runHandler();
         } catch (err) {
-          log.warn(err.stack);
+          this._safeLog('warning', err.stack);
         }
       }
     });