feature: support protobuf cloud events (#408)

Closes #391

Also improve efficiency of JSON decoding

Author: kevmoo

functions_framework/CHANGELOG.md

@@ -1,3 +1,7 @@
+## 0.4.3-wip
+
+- Support `application/protobuf` event encoding.
+
 ## 0.4.2
 
 - Requires Dart `3.0.0`.

functions_framework/lib/src/json_request_utils.dart

@@ -18,13 +18,14 @@ import 'package:gcp/gcp.dart';
 import 'package:http_parser/http_parser.dart';
 import 'package:shelf/shelf.dart';
 
-MediaType mediaTypeFromRequest(Request request) {
+MediaType mediaTypeFromRequest(Request request, {String? requiredMimeType}) {
   final contentType = request.headers[contentTypeHeader];
   if (contentType == null) {
     throw BadRequestException(400, '$contentTypeHeader header is required.');
   }
+  final MediaType value;
   try {
-    return MediaType.parse(contentType);
+    value = MediaType.parse(contentType);
   } catch (e, stack) {
     throw BadRequestException(
       400,
@@ -33,23 +34,23 @@ MediaType mediaTypeFromRequest(Request request) {
       innerStack: stack,
     );
   }
-}
 
-void mustBeJson(MediaType type) {
-  if (type.mimeType != jsonContentType) {
-    // https://github.com/GoogleCloudPlatform/functions-framework#http-status-codes
-    throw BadRequestException(
-      400,
-      'Unsupported encoding "${type.toString()}". '
-      'Only "$jsonContentType" is supported.',
-    );
+  if (requiredMimeType != null) {
+    if (value.mimeType != requiredMimeType) {
+      // https://github.com/GoogleCloudPlatform/functions-framework#http-status-codes
+      throw BadRequestException(
+        400,
+        'Unsupported encoding "${value.mimeType.toString()}". '
+        'Only "$requiredMimeType" is supported.',
+      );
+    }
   }
+  return value;
 }
 
-Future<Object?> decodeJson(Request request) async {
-  final content = await request.readAsString();
+Future<Object?> decodeJson(Stream<List<int>> data) async {
   try {
-    final value = jsonDecode(content);
+    final value = await utf8.decoder.bind(data).transform(json.decoder).single;
     return value;
   } on FormatException catch (e, stackTrace) {
     // https://github.com/GoogleCloudPlatform/functions-framework#http-status-codes
@@ -64,4 +65,44 @@ Future<Object?> decodeJson(Request request) async {
 
 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);
+    final supportedType = SupportedContentTypes.values.singleWhere(
+      (element) => element.value == type.mimeType,
+      orElse: () => throw BadRequestException(
+        400,
+        'Unsupported encoding "$type". '
+        'Supported types: '
+        '${SupportedContentTypes.values.map((e) => '"${e.value}"').join(', ')}',
+      ),
+    );
+
+    return (
+      mimeType: type,
+      data: await supportedType._decode(request.read()),
+    );
+  }
+
+  Future<Object?> _decode(
+    Stream<List<int>> data,
+  ) async =>
+      switch (this) {
+        json => await decodeJson(data),
+        protobuf => await data.fold<List<int>>(
+            <int>[],
+            (previous, element) => previous..addAll(element),
+          ),
+      };
+}
+
 const contentTypeHeader = 'Content-Type';

functions_framework/lib/src/targets/cloud_event_targets.dart

@@ -67,10 +67,8 @@ Future<CloudEvent> _eventFromRequest(Request request) =>
         : _decodeStructured(request);
 
 Future<CloudEvent> _decodeStructured(Request request) async {
-  final type = mediaTypeFromRequest(request);
-
-  mustBeJson(type);
-  var jsonObject = await decodeJson(request) as Map<String, dynamic>;
+  final type = mediaTypeFromRequest(request, requiredMimeType: jsonContentType);
+  var jsonObject = await decodeJson(request.read()) as Map<String, dynamic>;
 
   if (!jsonObject.containsKey('datacontenttype')) {
     jsonObject = {
@@ -86,15 +84,14 @@ const _cloudEventPrefix = 'ce-';
 const _clientEventPrefixLength = _cloudEventPrefix.length;
 
 Future<CloudEvent> _decodeBinary(Request request) async {
-  final type = mediaTypeFromRequest(request);
-  mustBeJson(type);
+  final data = await SupportedContentTypes.decode(request);
 
   final map = <String, Object?>{
     for (var e in request.headers.entries
         .where((element) => element.key.startsWith(_cloudEventPrefix)))
       e.key.substring(_clientEventPrefixLength): e.value,
-    'datacontenttype': type.toString(),
-    'data': await decodeJson(request),
+    'datacontenttype': data.mimeType.toString(),
+    'data': data.data,
   };
 
   return _decodeValidCloudEvent(map, 'binary-mode message');

functions_framework/lib/src/targets/json_targets.dart

@@ -28,9 +28,8 @@ abstract class _JsonFunctionTargetBase<RequestType> extends FunctionTarget {
   _JsonFunctionTargetBase(this._fromJson);
 
   Future<RequestType> _toRequestType(Request request) async {
-    final type = mediaTypeFromRequest(request);
-    mustBeJson(type);
-    final jsonObject = await decodeJson(request);
+    mediaTypeFromRequest(request, requiredMimeType: jsonContentType);
+    final jsonObject = await decodeJson(request.read());
     return _fromJson(jsonObject);
   }
 }

functions_framework/pubspec.yaml

@@ -1,5 +1,5 @@
 name: functions_framework
-version: 0.4.2
+version: 0.4.3-wip
 description: >-
   FaaS (Function as a service) framework for writing portable Dart functions
 repository: https://github.com/GoogleCloudPlatform/functions-framework-dart

integration_test/test/custom_type_test.dart

@@ -68,7 +68,7 @@ void main() {
       );
       await expectInvalid(
         response,
-        'Unsupported encoding "application/text; charset=utf-8". '
+        'Unsupported encoding "application/text". '
         'Only "application/json" is supported.',
       );
     });