Core: refactoring protobuf support (#413)

Encode data as Uint8List when possible
Add tests

Author: kevmoo

functions_framework/lib/src/json_request_utils.dart

@@ -13,6 +13,7 @@
 // limitations under the License.
 
 import 'dart:convert';
+import 'dart:typed_data';
 
 import 'package:gcp/gcp.dart';
 import 'package:http_parser/http_parser.dart';
@@ -49,6 +50,23 @@ MediaType mediaTypeFromRequest(Request request, {String? requiredMimeType}) {
 }
 
 extension RequestExt on Request {
+  Future<List<int>> readBytes() async {
+    final length = contentLength;
+    if (length == null) {
+      return await read().fold<List<int>>(
+        <int>[],
+        (previous, element) => previous..addAll(element),
+      );
+    }
+    final bytes = Uint8List(length);
+    var offset = 0;
+    await for (var bits in read()) {
+      bytes.setAll(offset, bits);
+      offset += bits.length;
+    }
+    return bytes;
+  }
+
   Future<Object?> decodeJson() async {
     try {
       final value = await (encoding ?? utf8)
@@ -67,22 +85,9 @@ extension RequestExt on Request {
       );
     }
   }
-}
-
-const jsonContentType = 'application/json';
-
-enum SupportedContentTypes {
-  json(jsonContentType),
-  protobuf('application/protobuf');
-
-  const SupportedContentTypes(this.value);
 
-  final String value;
-
-  static Future<({MediaType mimeType, Object? data})> decode(
-    Request request,
-  ) async {
-    final type = mediaTypeFromRequest(request);
+  Future<({MediaType mimeType, Object? data})> decode() async {
+    final type = mediaTypeFromRequest(this);
     final supportedType = SupportedContentTypes.values.singleWhere(
       (element) => element.value == type.mimeType,
       orElse: () => throw BadRequestException(
@@ -96,14 +101,22 @@ enum SupportedContentTypes {
     return (
       mimeType: type,
       data: switch (supportedType) {
-        json => await request.decodeJson(),
-        protobuf => await request.read().fold<List<int>>(
-            <int>[],
-            (previous, element) => previous..addAll(element),
-          ),
+        SupportedContentTypes.json => await decodeJson(),
+        SupportedContentTypes.protobuf => await readBytes(),
       },
     );
   }
 }
 
+const jsonContentType = 'application/json';
+
+enum SupportedContentTypes {
+  json(jsonContentType),
+  protobuf('application/protobuf');
+
+  const SupportedContentTypes(this.value);
+
+  final String value;
+}
+
 const contentTypeHeader = 'Content-Type';

functions_framework/lib/src/targets/cloud_event_targets.dart

@@ -84,7 +84,7 @@ const _cloudEventPrefix = 'ce-';
 const _clientEventPrefixLength = _cloudEventPrefix.length;
 
 Future<CloudEvent> _decodeBinary(Request request) async {
-  final data = await SupportedContentTypes.decode(request);
+  final data = await request.decode();
 
   final map = <String, Object?>{
     for (var e in request.headers.entries

integration_test/bin/server.dart

@@ -139,6 +139,18 @@ void basicCloudEventHandler(CloudEvent event, RequestContext context) {
   stderr.writeln(encodeJsonPretty(event));
 }
 
+@CloudFunction()
+void protoEventHandler(CloudEvent event, RequestContext context) {
+  context.logger.info('event subject: ${event.subject}');
+
+  context.logger.debug(context.request.headers);
+
+  context.responseHeaders['x-data-runtime-types'] =
+      event.data.runtimeType.toString();
+
+  stderr.writeln(encodeJsonPretty(event));
+}
+
 final _helloWorldBytes = utf8.encode('Hello, World!');
 
 const _contentTypeHeader = 'Content-Type';

integration_test/lib/functions.dart

@@ -98,6 +98,67 @@ void main() {
       );
     });
 
+    test('valid proto input', () async {
+      final proc = await _hostBasicEventHandler('protoEventHandler');
+
+      const subject = 'documents/users/ghXNtePIFmdDOBH3iEMH';
+      final response = await _makeRequest(
+        _protobytes,
+        {
+          'ce-id': '785865c0-2b16-439b-ad68-f9672343863a',
+          'ce-source':
+              '//firestore.googleapis.com/projects/dart-redirector/databases/(default)',
+          'ce-specversion': '1.0',
+          'ce-type': 'google.cloud.firestore.document.v1.updated',
+          'Content-Type': 'application/protobuf',
+          'ce-dataschema':
+              'https://github.com/googleapis/google-cloudevents/blob/main/proto/google/events/cloud/firestore/v1/data.proto',
+          'ce-subject': subject,
+          'ce-time': '2023-06-21T12:21:25.413855Z',
+        },
+      );
+      expect(response.statusCode, 200);
+      expect(response.body, isEmpty);
+      expect(
+        response.headers,
+        allOf(
+          containsTextPlainHeader,
+          containsPair('x-data-runtime-types', 'Uint8List'),
+        ),
+      );
+      await expectLater(
+        proc.stdout,
+        emitsInOrder(
+          [
+            startsWith('INFO: event subject: $subject'),
+            startsWith('DEBUG:'),
+          ],
+        ),
+      );
+
+      await finishServerTest(
+        proc,
+        requestOutput: endsWith('POST    [200] /'),
+      );
+
+      final stderrOutput = await proc.stderrStream().join('\n');
+      final json = jsonDecode(stderrOutput) as Map<String, dynamic>;
+
+      expect(json, {
+        'id': '785865c0-2b16-439b-ad68-f9672343863a',
+        'source':
+            '//firestore.googleapis.com/projects/dart-redirector/databases/(default)',
+        'specversion': '1.0',
+        'type': 'google.cloud.firestore.document.v1.updated',
+        'datacontenttype': 'application/protobuf',
+        'data': _protobytes,
+        'dataschema':
+            'https://github.com/googleapis/google-cloudevents/blob/main/proto/google/events/cloud/firestore/v1/data.proto',
+        'subject': 'documents/users/ghXNtePIFmdDOBH3iEMH',
+        'time': '2023-06-21T12:21:25.413855Z',
+      });
+    });
+
     test('bad format of core header: time', () async {
       final stderrOutput = await _makeBadRequest(
         _pubSubJsonString,
@@ -311,7 +372,7 @@ Future<String> _makeBadRequest(
   return stderrOutput;
 }
 
-Future<Response> _makeRequest(String body, Map<String, String> headers) async {
+Future<Response> _makeRequest(Object? body, Map<String, String> headers) async {
   final requestUrl = 'http://localhost:$autoPort/';
 
   final response = await post(
@@ -322,15 +383,26 @@ Future<Response> _makeRequest(String body, Map<String, String> headers) async {
   return response;
 }
 
-Future<TestProcess> _hostBasicEventHandler() async {
+Future<TestProcess> _hostBasicEventHandler([
+  String name = 'basicCloudEventHandler',
+]) async {
   final proc = await startServerTest(
     arguments: [
       '--target',
-      'basicCloudEventHandler',
+      name,
       '--signature-type',
       'cloudevent',
     ],
     expectedListeningPort: 0,
   );
   return proc;
 }
+
+/// This is a Firestore update record in protobuf
+final _protobytes = base64Decode(
+  'CoIBClFwcm9qZWN0cy9kYXJ0LXJlZGlyZWN0b3IvZGF0YWJhc2VzLyhkZWZhdWx0KS9kb2N1bWVu'
+  'dHMvdXNlcnMvZ2hYTnRlUElGbWRET0JIM2lFTUgSEQoEbmFtZRIJigEGbHVjaWE0GgwI6+KupAYQ'
+  'iMa2qgIiDAjF1sukBhCY2qvFARKCAQpRcHJvamVjdHMvZGFydC1yZWRpcmVjdG9yL2RhdGFiYXNl'
+  'cy8oZGVmYXVsdCkvZG9jdW1lbnRzL3VzZXJzL2doWE50ZVBJRm1kRE9CSDNpRU1IEhEKBG5hbWUS'
+  'CYoBBmx1Y2lhMxoMCOvirqQGEIjGtqoCIgwI8o60pAYQmJa6igMaBgoEbmFtZQ==',
+);