GoogleCloudPlatform · coryan · Feb 19, 2021 · Feb 18, 2021 · Feb 19, 2021 · Feb 19, 2021
@@ -47,6 +47,8 @@ add_library(
     internal/parse_cloud_event_http.h
     internal/parse_cloud_event_json.cc
     internal/parse_cloud_event_json.h
+    internal/parse_cloud_event_storage.cc
+    internal/parse_cloud_event_storage.h
     internal/parse_options.cc
     internal/parse_options.h
     internal/setenv.cc
@@ -95,6 +97,7 @@ if (BUILD_TESTING)
         internal/framework_impl_test.cc
         internal/parse_cloud_event_http_test.cc
         internal/parse_cloud_event_json_test.cc
+        internal/parse_cloud_event_storage_test.cc
         internal/parse_options_test.cc
         internal/wrap_request_test.cc
         version_test.cc)

@@ -14,6 +14,7 @@
 
 #include "google/cloud/functions/internal/parse_cloud_event_http.h"
 #include "google/cloud/functions/internal/parse_cloud_event_json.h"
+#include "google/cloud/functions/internal/parse_cloud_event_storage.h"
 
 namespace google::cloud::functions_internal {
 inline namespace FUNCTIONS_FRAMEWORK_CPP_NS {
@@ -60,7 +61,7 @@ functions::CloudEvent ParseCloudEventHttpBinary(BeastRequest const& request) {
 std::vector<functions::CloudEvent> ParseCloudEventHttp(
     BeastRequest const& request) {
   if (request.count("content-type") == 0) {
-    return {ParseCloudEventHttpBinary(request)};
+    return {ParseCloudEventStorage(ParseCloudEventHttpBinary(request))};
   }
   auto content_type = request["content-type"];
   if (content_type.rfind("application/cloudevents-batch+json", 0) == 0) {
@@ -69,7 +70,7 @@ std::vector<functions::CloudEvent> ParseCloudEventHttp(
   if (content_type.rfind("application/cloudevents+json", 0) == 0) {
     return {ParseCloudEventJson(request.body())};
   }
-  return {ParseCloudEventHttpBinary(request)};
+  return {ParseCloudEventStorage(ParseCloudEventHttpBinary(request))};
 }
 
 }  // namespace FUNCTIONS_FRAMEWORK_CPP_NS

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 #include "google/cloud/functions/internal/parse_cloud_event_http.h"
+#include <cppcodec/base64_rfc4648.hpp>
 #include <gmock/gmock.h>
+#include <nlohmann/json.hpp>
 #include <algorithm>
 #include <iterator>
 
@@ -140,6 +142,69 @@ TEST(ParseCloudEventHttp, Json) {
   EXPECT_EQ(ce.spec_version(), functions::CloudEvent::kDefaultSpecVersion);
 }
 
+TEST(ParseCloudEventJson, EmulateStorage) {
+  auto const data = nlohmann::json::parse(R"js({
+    "bucket": "some-bucket",
+    "contentType": "text/plain",
+    "crc32c": "rTVTeQ==",
+    "etag": "CNHZkbuF/ugCEAE=",
+    "generation": "1587627537231057",
+    "id": "some-bucket/folder/Test.cs/1587627537231057",
+    "kind": "storage#object",
+    "md5Hash": "kF8MuJ5+CTJxvyhHS1xzRg==",
+    "mediaLink": "https://www.googleapis.com/download/storage/v1/b/some-bucket/o/folder%2FTest.cs?generation=1587627537231057\u0026alt=media",
+    "metageneration": "1",
+    "name": "folder/Test.cs",
+    "selfLink": "https://www.googleapis.com/storage/v1/b/some-bucket/o/folder/Test.cs",
+    "size": "352",
+    "storageClass": "MULTI_REGIONAL",
+    "timeCreated": "2020-04-23T07:38:57.230Z",
+    "timeStorageClassUpdated": "2020-04-23T07:38:57.230Z",
+    "updated": "2020-04-23T07:38:57.230Z"
+  })js");
+  auto const data_base64 = cppcodec::base64_rfc4648::encode(data.dump());
+  auto const attributes = nlohmann::json{
+      {"notificationConfig",
+       "projects/_/buckets/some-bucket/notificationConfigs/3"},
+      {"eventType", "OBJECT_FINALIZE"},
+      {"payloadFormat", "JSON_API_V1"},
+      {"bucketId", "some-bucket"},
+      {"objectId", "folder/Test.cs"},
+      {"objectGeneration", "1587627537231057"},
+  };
+  auto const payload = nlohmann::json{{
+      "message",
+      nlohmann::json{
+          {"attributes", attributes},
+          {"data", data_base64},
+      },
+  }};
+
+  auto request = BeastRequest();
+  request.insert("ce-type", "google.cloud.pubsub.topic.v1.messagePublished");
+  request.insert(
+      "ce-source",
+      "//pubsub.googleapis.com/projects/sample-project/topics/storage");
+  request.insert("ce-id", "A234-1234-1234");
+  request.insert("content-type", "application/json");
+  request.body() = payload.dump();
+  request.prepare_payload();
+
+  auto events = ParseCloudEventHttp(request);
+  ASSERT_THAT(events.size(), 1);
+  auto const& ce = events[0];
+  EXPECT_EQ(ce.id(), "A234-1234-1234");
+  EXPECT_EQ(ce.source(),
+            "//storage.googleapis.com/projects/_/buckets/some-bucket");
+  EXPECT_EQ(ce.type(), "google.cloud.storage.object.v1.finalized");
+  EXPECT_EQ(ce.spec_version(), functions::CloudEvent::kDefaultSpecVersion);
+  ASSERT_EQ(ce.data_content_type().value_or(""), "application/json");
+  auto const actual_storage_data =
+      nlohmann::json::parse(ce.data().value_or("{}"));
+  auto const delta = nlohmann::json::diff(data, actual_storage_data);
+  EXPECT_EQ(data, actual_storage_data) << "delta=" << delta;
+}
+
 TEST(ParseCloudEventHttp, JsonBatch) {
   auto constexpr kText = R"js([
   {

@@ -14,6 +14,7 @@
 
 #include "google/cloud/functions/internal/parse_cloud_event_json.h"
 #include "google/cloud/functions/internal/base64_decode.h"
+#include "google/cloud/functions/internal/parse_cloud_event_storage.h"
 #include <nlohmann/json.hpp>
 #include <algorithm>
 
@@ -59,7 +60,7 @@ functions::CloudEvent ParseCloudEventJson(nlohmann::json const& json) {
 /// Parse @p json_string as a Cloud Event
 functions::CloudEvent ParseCloudEventJson(std::string_view json_string) {
   auto json = nlohmann::json::parse(json_string);
-  return ParseCloudEventJson(json);
+  return ParseCloudEventStorage(ParseCloudEventJson(json));
 }
 
 std::vector<functions::CloudEvent> ParseCloudEventJsonBatch(

@@ -13,7 +13,9 @@
 // limitations under the License.
 
 #include "google/cloud/functions/internal/parse_cloud_event_json.h"
+#include <cppcodec/base64_rfc4648.hpp>
 #include <gmock/gmock.h>
+#include <nlohmann/json.hpp>
 #include <algorithm>
 #include <iterator>
 
@@ -225,6 +227,64 @@ TEST(ParseCloudEventJson, BatchInvalid) {
                std::exception);
 }
 
+TEST(ParseCloudEventJson, EmulateStorage) {
+  auto const data = nlohmann::json::parse(R"js({
+    "bucket": "some-bucket",
+    "contentType": "text/plain",
+    "crc32c": "rTVTeQ==",
+    "etag": "CNHZkbuF/ugCEAE=",
+    "generation": "1587627537231057",
+    "id": "some-bucket/folder/Test.cs/1587627537231057",
+    "kind": "storage#object",
+    "md5Hash": "kF8MuJ5+CTJxvyhHS1xzRg==",
+    "mediaLink": "https://www.googleapis.com/download/storage/v1/b/some-bucket/o/folder%2FTest.cs?generation=1587627537231057\u0026alt=media",
+    "metageneration": "1",
+    "name": "folder/Test.cs",
+    "selfLink": "https://www.googleapis.com/storage/v1/b/some-bucket/o/folder/Test.cs",
+    "size": "352",
+    "storageClass": "MULTI_REGIONAL",
+    "timeCreated": "2020-04-23T07:38:57.230Z",
+    "timeStorageClassUpdated": "2020-04-23T07:38:57.230Z",
+    "updated": "2020-04-23T07:38:57.230Z"
+  })js");
+  auto const data_base64 = cppcodec::base64_rfc4648::encode(data.dump());
+  auto const attributes = nlohmann::json{
+      {"notificationConfig",
+       "projects/_/buckets/some-bucket/notificationConfigs/3"},
+      {"eventType", "OBJECT_FINALIZE"},
+      {"payloadFormat", "JSON_API_V1"},
+      {"bucketId", "some-bucket"},
+      {"objectId", "folder/Test.cs"},
+      {"objectGeneration", "1587627537231057"},
+  };
+  auto const payload = nlohmann::json{{
+      "message",
+      nlohmann::json{
+          {"attributes", attributes},
+          {"data", data_base64},
+      },
+  }};
+
+  auto event = nlohmann::json{
+      {"specversion", "1.0"},
+      {"type", "google.cloud.pubsub.topic.v1.messagePublished"},
+      {"source",
+       "//pubsub.googleapis.com/projects/sample-project/topics/storage"},
+      {"id", "aaaaaa-1111-bbbb-2222-cccccccccccc"},
+      {"time", "2020-09-29T11:32:00.000Z"},
+      {"datacontenttype", "application/json"},
+      {"data", payload},
+  };
+
+  auto const ce = ParseCloudEventJson(event.dump());
+  EXPECT_EQ(ce.type(), "google.cloud.storage.object.v1.finalized");
+  ASSERT_EQ(ce.data_content_type().value_or(""), "application/json");
+  auto const actual_storage_data =
+      nlohmann::json::parse(ce.data().value_or("{}"));
+  auto const delta = nlohmann::json::diff(data, actual_storage_data);
+  EXPECT_EQ(data, actual_storage_data) << "delta=" << delta;
+}
+
 }  // namespace
 }  // namespace FUNCTIONS_FRAMEWORK_CPP_NS
 }  // namespace google::cloud::functions_internal
@@ -0,0 +1,68 @@
+// Copyright 2021 Google LLC
+//
+// 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.
+
+#include "google/cloud/functions/internal/parse_cloud_event_storage.h"
+#include "google/cloud/functions/internal/base64_decode.h"
+#include <nlohmann/json.hpp>
+
+namespace google::cloud::functions_internal {
+inline namespace FUNCTIONS_FRAMEWORK_CPP_NS {
+
+functions::CloudEvent ParseCloudEventStorage(functions::CloudEvent e) {
+  if (e.type() != "google.cloud.pubsub.topic.v1.messagePublished") return e;
+  if (e.data_content_type().value_or("") != "application/json") return e;
+
+  // If the event looks like a storage event, reparse it and return that event
+  // instead.
+  auto const payload = nlohmann::json::parse(e.data().value_or("{}"));
+  if (payload.count("message") == 0) return e;
+  auto const& message = payload.at("message");
+  if (message.count("attributes") == 0 || message.count("data") == 0) return e;
+  auto const& attributes = message.at("attributes");
+  char const* required_attributes[] = {
+      "notificationConfig", "eventType", "payloadFormat",
+      "bucketId",           "objectId",  "objectGeneration",
+  };
+  auto const has_all_attributes = std::all_of(
+      std::begin(required_attributes), std::end(required_attributes),
+      [&attributes](char const* a) { return attributes.count(a) != 0; });
+  if (!has_all_attributes) return e;
+  if (attributes.value("payloadFormat", "") != "JSON_API_V1") return e;
+  static auto const kMessageTypeMappings =
+      std::unordered_map<std::string, std::string>{
+          {"OBJECT_FINALIZE", "google.cloud.storage.object.v1.finalized"},
+          {"OBJECT_METADATA_UPDATE",
+           "google.cloud.storage.object.v1.metadataUpdated"},
+          {"OBJECT_DELETE", "google.cloud.storage.object.v1.deleted"},
+          {"OBJECT_ARCHIVE", "google.cloud.storage.object.v1.archived"},
+      };
+  auto mapped = kMessageTypeMappings.find(attributes.value("eventType", ""));
+  if (mapped == kMessageTypeMappings.end()) return e;
+
+  auto source = "//storage.googleapis.com/projects/_/buckets/" +
+                attributes.value("bucketId", "");
+  auto const& event_type = mapped->second;
+  auto event = functions::CloudEvent(e.id(), std::move(source), event_type,
+                                     e.spec_version());
+  event.set_data_content_type("application/json");
+  event.set_data_schema("google.events.cloud.storage.v1.StorageObjectData");
+  event.set_subject("objects/" + attributes.value("objectId", ""));
+  if (e.time().has_value()) event.set_time(e.time().value());
+  event.set_data(Base64Decode(message.value("data", "")));
+
+  return event;
+}
+
+}  // namespace FUNCTIONS_FRAMEWORK_CPP_NS
+}  // namespace google::cloud::functions_internal
@@ -0,0 +1,30 @@
+// Copyright 2021 Google LLC
+//
+// 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.
+
+#ifndef FUNCTIONS_FRAMEWORK_CPP_GOOGLE_CLOUD_FUNCTIONS_INTERNAL_PARSE_CLOUD_EVENT_STORAGE_H
+#define FUNCTIONS_FRAMEWORK_CPP_GOOGLE_CLOUD_FUNCTIONS_INTERNAL_PARSE_CLOUD_EVENT_STORAGE_H
+
+#include "google/cloud/functions/cloud_event.h"
+#include "google/cloud/functions/version.h"
+
+namespace google::cloud::functions_internal {
+inline namespace FUNCTIONS_FRAMEWORK_CPP_NS {
+
+/// Parse @p e as a Cloud Storage event if possible, otherwise return @p e.
+functions::CloudEvent ParseCloudEventStorage(functions::CloudEvent e);
+
+}  // namespace FUNCTIONS_FRAMEWORK_CPP_NS
+}  // namespace google::cloud::functions_internal
+
+#endif  // FUNCTIONS_FRAMEWORK_CPP_GOOGLE_CLOUD_FUNCTIONS_INTERNAL_PARSE_CLOUD_EVENT_STORAGE_H