Add "shim" (mini-emulator) for GCF functions

functions/helloworld/package.json

@@ -24,14 +24,16 @@
     "safe-buffer": "5.1.1"
   },
   "devDependencies": {
-    "@google-cloud/nodejs-repo-tools": "2.2.5",
+    "@google-cloud/nodejs-repo-tools": "^2.2.5",
     "@google-cloud/pubsub": "^0.17.0",
     "@google-cloud/storage": "^1.5.0",
     "ava": "0.25.0",
+    "express": "^4.16.3",
     "proxyquire": "2.0.1",
     "sinon": "4.4.8",
     "supertest": "^3.0.0",
-    "uuid": "^3.1.0"
+    "uuid": "^3.1.0",
+    "yargs": "^11.0.0"
   },
   "cloud-repo-tools": {
     "requiresKeyFile": true,

functions/helloworld/shim.js

@@ -0,0 +1,156 @@
+/**
+ * Copyright 2018, Google, Inc.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+// [START functions_testing_shim_http]
+// [START functions_testing_shim_storage]
+// [START functions_testing_shim_pubsub]
+// Import tested code + dependencies
+const gcfCode = require('./index.js');
+// [END functions_testing_shim_http]
+// [END functions_testing_shim_storage]
+// [END functions_testing_shim_pubsub]
+
+const httpShim = (PORT) => {
+  // [START functions_testing_shim_http]
+  const express = require('express');
+
+  // TODO(developer): specify the port to use
+  // const PORT = 3000;
+
+  // Start local HTTP server
+  const app = express();
+  const server = require(`http`).createServer(app);
+  server.on('connection', socket => socket.unref());
+  server.listen(PORT);
+
+  // Register HTTP handlers
+  Object.keys(gcfCode).forEach(gcfFn => {
+    // Handle a single HTTP request
+    const handler = (req, res) => {
+      gcfCode[gcfFn](req, res);
+      server.close();
+    };
+
+    app.get(`/${gcfFn}`, handler);
+    app.post(`/${gcfFn}`, handler);
+  });
+  // [END functions_testing_shim_http]
+};
+
+const pubsubShim = (gcfFn, topicName, subscriptionName) => {
+  // Import dependencies (in function, to avoid cluttering other samples)
+  // [START functions_testing_shim_pubsub]
+  const Pubsub = require('@google-cloud/pubsub');
+  const pubsub = Pubsub();
+
+  // TODO(developer): specify a function to test
+  // const gcfFn = gcfCode.YOUR_FUNCTION;
+
+  // TODO(developer): specify an existing topic and subscription to use
+  // const topicName = process.env.TOPIC || 'YOUR_TOPIC';
+  // const subscriptionName = process.env.SUBSCRIPTION || 'YOUR_SUBSCRIPTION';
+
+  // Subscribe to Pub/Sub topic
+  const subscription = pubsub.topic(topicName).subscription(subscriptionName);
+
+  // Handle a single Pub/Sub message
+  const messageHandler = (msg) => {
+    gcfFn({ data: msg }, () => {
+      msg.ack();
+      subscription.removeListener(`message`, messageHandler);
+    });
+  };
+  subscription.on(`message`, messageHandler);
+  // [END functions_testing_shim_pubsub]
+};
+
+const storageShim = (gcfFn, bucketName, topicName, subscriptionName) => {
+  // Import dependencies (in function, to avoid cluttering other samples)
+  // [START functions_testing_shim_storage]
+  const Pubsub = require('@google-cloud/pubsub');
+  const Storage = require(`@google-cloud/storage`);
+  const pubsub = Pubsub();
+  const storage = Storage();
+
+  // TODO(developer): specify a function to test
+  // const gcfFn = gcfCode.YOUR_FUNCTION;
+
+  // TODO(developer): specify a Cloud Storage bucket to monitor
+  // const bucketName = 'YOUR_GCS_BUCKET'
+
+  // TODO(developer): specify an existing topic and subscription to use
+  // const topicName = process.env.TOPIC || 'YOUR_TOPIC';
+  // const subscriptionName = process.env.SUBSCRIPTION || 'YOUR_SUBSCRIPTION';
+
+  // Create notification on target bucket
+  // Further info: https://cloud.google.com/storage/docs/reporting-changes
+  const bucket = storage.bucket(bucketName);
+  return bucket.createNotification(topicName)
+    .then(data => data[0])
+    .then((notification) => new Promise(resolve => {
+      // Subscribe to Pub/Sub topic
+      const subscription = pubsub
+        .topic(topicName)
+        .subscription(subscriptionName);
+
+      // Handle a single Pub/Sub message
+      const messageHandler = (msg) => {
+        console.error('DATA', Buffer.from(msg.data, 'base64').toString());
+        const data = JSON.parse(Buffer.from(msg.data, 'base64').toString());
+        gcfFn({ data: data }, () => {
+          msg.ack();
+          subscription.removeListener(`message`, messageHandler);
+          resolve(notification);
+        });
+      };
+      subscription.on(`message`, messageHandler);
+    }))
+    .then(notification => notification.delete()); // Delete notification
+  // [END functions_testing_shim_storage]
+};
+
+require(`yargs`) // eslint-disable-line
+  .demandCommand(1)
+  .command(
+    'http <port>',
+    'HTTP-triggered-function shim',
+    {},
+    opts => httpShim(opts.port)
+  )
+  .command(
+    'pubsub <functionName> <topic> <subscription>',
+    'PubSub-triggered-function shim',
+    {},
+    opts => pubsubShim(
+      gcfCode[opts.functionName],
+      opts.topic,
+      opts.subscription
+    )
+  )
+  .command(
+    'storage <functionName> <bucket> <topic> <subscription>',
+    'Storage-triggered-function shim',
+    {},
+    opts => storageShim(
+      gcfCode[opts.functionName],
+      opts.bucket,
+      opts.topic,
+      opts.subscription
+    )
+  )
+  .wrap(120)
+  .help()
+  .strict()
+  .argv;

functions/helloworld/test/shim.test.js

@@ -0,0 +1,113 @@
+/**
+ * Copyright 2018, Google, Inc.
+ * Licensed under the Apache License, Version 2.0 (the "License");
+ * you may not use this file except in compliance with the License.
+ * You may obtain a copy of the License at
+ *
+ *    http://www.apache.org/licenses/LICENSE-2.0
+ *
+ * Unless required by applicable law or agreed to in writing, software
+ * distributed under the License is distributed on an "AS IS" BASIS,
+ * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+ * See the License for the specific language governing permissions and
+ * limitations under the License.
+ */
+
+const test = require(`ava`);
+const Supertest = require(`supertest`);
+const tools = require(`@google-cloud/nodejs-repo-tools`);
+const Pubsub = require(`@google-cloud/pubsub`);
+const pubsub = Pubsub();
+const Storage = require(`@google-cloud/storage`);
+const storage = Storage();
+const uuid = require(`uuid`);
+const path = require(`path`);
+
+const PORT = process.env.PORT || 3000;
+const supertest = Supertest(`http://localhost:${PORT}`);
+const topicName = `gcf-shim-topic-${uuid.v4()}`;
+const subscriptionName = `gcf-shim-sub-${uuid.v4()}`;
+const bucketName = `gcf-shim-bucket-${uuid.v4()}`;
+
+const cmd = `node shim.js`;
+const cwd = path.join(__dirname, `..`);
+
+const bucket = storage.bucket(bucketName);
+let topic;
+let subscription;
+
+// Helper function
+const delay = (ms) => new Promise(resolve => setTimeout(resolve, ms));
+
+// Create test resources
+test.before(t => {
+  return bucket.create()
+    .then(data => {
+      return pubsub.createTopic(topicName);
+    })
+    .then(data => {
+      topic = data[0];
+      return topic.createSubscription(subscriptionName);
+    })
+    .then(data => {
+      subscription = data[0];
+      return delay(500);
+    })
+    .catch(console.error);
+});
+
+// Tear down test resources
+test.after.always(t => {
+  return bucket.deleteFiles({ force: true })
+    .then(subscription.delete())
+    .then(topic.delete())
+    .then(() => {
+      try {
+        return bucket.delete();
+      } catch (ex) {
+        return bucket.delete();
+      }
+    })
+    .catch(console.error);
+});
+
+test(`shim: should handle HTTP`, async t => {
+  const shim = tools.runAsync(`${cmd} http ${PORT}`, cwd);
+  await delay(1000)
+    .then(() => {
+      // Send HTTP request
+      return supertest
+        .get(`/helloGET`)
+        .expect(200)
+        .expect((response) => {
+          t.is(response.text, 'Hello World!');
+        });
+    })
+    .then(() => { return shim; }); // Stop shim
+});
+
+test(`shim: should handle PubSub`, async t => {
+  const name = uuid.v4();
+  const publisher = topic.publisher();
+  const shim = tools.runAsync(`${cmd} pubsub helloPubSub "${topicName}" "${subscriptionName}"`, cwd);
+
+  // Publish to topic
+  await publisher.publish(Buffer.from(name))
+    .then(delay(1000))
+    .then(() => { return shim; })
+    .then(output => {
+      t.true(output.includes(`Hello, ${name}!`));
+    });
+});
+
+test(`shim: should handle GCS`, async t => {
+  const shim = tools.runAsync(`${cmd} storage helloGCS "${bucketName}" "${topicName}" "${subscriptionName}"`, cwd);
+
+  // Upload file to bucket
+  await delay(10000)
+    .then(() => { return bucket.upload(path.join(cwd, `test/test.txt`)); })
+    .then(() => { return shim; })
+    .then(output => {
+      t.true(output.includes(`File test.txt uploaded.`));
+    });
+});