feat: In CE binary mode, support protobuf binary wireformat

Currently this expects a datacontenttype attribute with a media type
of "application/protobuf".

src/Google.Events.Protobuf.Tests/ProtobufJsonCloudEventFormatterTest.cs

@@ -14,6 +14,7 @@
 
 using CloudNative.CloudEvents;
 using Google.Events.Protobuf.Cloud.Storage.V1;
+using Google.Protobuf;
 using Newtonsoft.Json;
 using Newtonsoft.Json.Linq;
 using System;
@@ -84,6 +85,21 @@ public void DecodeBinary_NoData()
             Assert.Null(cloudEvent.Data);
         }
 
+        [Fact]
+        public void DecodeBinary_ProtobufWireFormat()
+        {
+            var cloudEvent = CreateSampleEvent();
+            var expectedData = cloudEvent.Data;
+            var binaryData = ((IMessage) expectedData).ToByteArray();
+            cloudEvent.Data = null;
+            cloudEvent.DataContentType = "application/protobuf";
+
+            var converter = new ProtobufJsonCloudEventFormatter<StorageObjectData>();
+            converter.DecodeBinaryModeEventData(binaryData, cloudEvent);
+            var actualData = cloudEvent.Data;
+            Assert.Equal(expectedData, actualData);
+        }
+
         [Fact]
         public void DecodeBatch()
         {

src/Google.Events.Protobuf/ProtobufJsonCloudEventFormatter.cs

@@ -28,19 +28,40 @@ namespace Google.Events.Protobuf
     /// <summary>
     /// <see cref="CloudEventFormatter"/> which is able to serialize protobuf messages in JSON.
     /// (Note that this does not implement the protobuf CloudEvent format, which is separate.)
+    /// Additionally, when deserializing binary mode events, if the data content type has a media type of "application/protobuf",
+    /// this will deserialize the body as protobuf binary data.
     /// </summary>
     /// <typeparam name="T">The type of message to serialize/deserialize</typeparam>
     public class ProtobufJsonCloudEventFormatter<T> : CloudEventFormatter where T : class, IMessage<T>, new()
     {
+        private const string BinaryProtobufMediaType = "application/protobuf";
+
         private static readonly JsonParser s_jsonParser = new JsonParser(JsonParser.Settings.Default.WithIgnoreUnknownFields(true));
 
         // Note: although we delegate to a JsonEventFormatter (via StructuredEventFormatter) and *could* just derive from JsonEventFormatter,
         // that would provide less flexibility for the future.
         private static readonly CloudEventFormatter s_structuredEventFormatter = new StructuredEventFormatter();
 
         /// <inheritdoc />
-        public override void DecodeBinaryModeEventData(ReadOnlyMemory<byte> body, CloudEvent cloudEvent) =>
-            cloudEvent.Data = body.Length == 0 ? null : s_jsonParser.Parse<T>(new StreamReader(BinaryDataUtilities.AsStream(body)));
+        public override void DecodeBinaryModeEventData(ReadOnlyMemory<byte> body, CloudEvent cloudEvent)
+        {
+            if (body.Length == 0)
+            {
+                cloudEvent.Data = null;
+                return;
+            }
+            ContentType dataContentType = new ContentType(cloudEvent.DataContentType ?? "application/json");
+            cloudEvent.Data =
+                dataContentType.MediaType == BinaryProtobufMediaType ? ParseBinaryProtobuf(body)
+                : s_jsonParser.Parse<T>(new StreamReader(BinaryDataUtilities.AsStream(body)));
+        }
+
+        private T ParseBinaryProtobuf(ReadOnlyMemory<byte> body)
+        {
+            var message = new T();
+            message.MergeFrom(body.Span);
+            return message;
+        }
 
         // TODO: Handle other structured modes beyond JSON?