Add ToCloudEventBatch to HttpListenerExtensions

Fixes #166

Signed-off-by: Jon Skeet <jonskeet@google.com>

src/CloudNative.CloudEvents/Http/HttpListenerExtensions.cs

@@ -212,6 +212,77 @@ namespace CloudNative.CloudEvents.Http
             }
         }
 
+        /// <summary>
+        /// Converts this HTTP request message into a CloudEvent batch.
+        /// </summary>
+        /// <param name="httpListenerRequest">The HTTP request to convert. Must not be null.</param>
+        /// <param name="formatter">The event formatter to use to parse the CloudEvents. Must not be null.</param>
+        /// <param name="extensionAttributes">The extension attributes to use when parsing the CloudEvents. May be null.</param>
+        /// <returns>The decoded batch of CloudEvents.</returns>
+        public static Task<IReadOnlyList<CloudEvent>> ToCloudEventBatchAsync(
+            this HttpListenerRequest httpListenerRequest,
+            CloudEventFormatter formatter,
+            params CloudEventAttribute[] extensionAttributes) =>
+            ToCloudEventBatchAsync(httpListenerRequest, formatter, (IEnumerable<CloudEventAttribute>)extensionAttributes);
+
+        /// <summary>
+        /// Converts this HTTP request message into a CloudEvent batch.
+        /// </summary>
+        /// <param name="httpListenerRequest">The HTTP request to convert. Must not be null.</param>
+        /// <param name="formatter">The event formatter to use to parse the CloudEvent. Must not be null.</param>
+        /// <param name="extensionAttributes">The extension attributes to use when parsing the CloudEvent. May be null.</param>
+        /// <returns>The decoded batch of CloudEvents.</returns>
+        public static async Task<IReadOnlyList<CloudEvent>> ToCloudEventBatchAsync(
+            this HttpListenerRequest httpListenerRequest,
+            CloudEventFormatter formatter,
+            IEnumerable<CloudEventAttribute> extensionAttributes) =>
+            await ToCloudEventBatchInternalAsync(httpListenerRequest, formatter, extensionAttributes, async: true).ConfigureAwait(false);
+
+        /// <summary>
+        /// Converts this HTTP request message into a CloudEvent batch.
+        /// </summary>
+        /// <param name="httpListenerRequest">The HTTP request to convert. Must not be null.</param>
+        /// <param name="formatter">The event formatter to use to parse the CloudEvents. Must not be null.</param>
+        /// <param name="extensionAttributes">The extension attributes to use when parsing the CloudEvents. May be null.</param>
+        /// <returns>The decoded batch of CloudEvents.</returns>
+        public static IReadOnlyList<CloudEvent> ToCloudEventBatch(
+            this HttpListenerRequest httpListenerRequest,
+            CloudEventFormatter formatter,
+            params CloudEventAttribute[] extensionAttributes) =>
+            ToCloudEventBatch(httpListenerRequest, formatter, (IEnumerable<CloudEventAttribute>) extensionAttributes);
+
+        /// <summary>
+        /// Converts this HTTP request message into a CloudEvent batch.
+        /// </summary>
+        /// <param name="httpListenerRequest">The HTTP request to convert. Must not be null.</param>
+        /// <param name="formatter">The event formatter to use to parse the CloudEvents. Must not be null.</param>
+        /// <param name="extensionAttributes">The extension attributes to use when parsing the CloudEvents. May be null.</param>
+        /// <returns>The decoded batch of CloudEvents.</returns>
+        public static IReadOnlyList<CloudEvent> ToCloudEventBatch(
+            this HttpListenerRequest httpListenerRequest,
+            CloudEventFormatter formatter,
+            IEnumerable<CloudEventAttribute> extensionAttributes) =>
+            ToCloudEventBatchInternalAsync(httpListenerRequest, formatter, extensionAttributes, async: false).GetAwaiter().GetResult();
+        
+        private async static Task<IReadOnlyList<CloudEvent>> ToCloudEventBatchInternalAsync(HttpListenerRequest httpListenerRequest,
+            CloudEventFormatter formatter, IEnumerable<CloudEventAttribute> extensionAttributes, bool async)
+        {
+            Validation.CheckNotNull(httpListenerRequest, nameof(httpListenerRequest));
+            Validation.CheckNotNull(formatter, nameof(formatter));
+
+            if (HasCloudEventsBatchContentType(httpListenerRequest))
+            {
+                var contentType = MimeUtilities.CreateContentTypeOrNull(httpListenerRequest.ContentType);
+                return async
+                    ? await formatter.DecodeBatchModeMessageAsync(httpListenerRequest.InputStream, contentType, extensionAttributes).ConfigureAwait(false)
+                    : formatter.DecodeBatchModeMessage(httpListenerRequest.InputStream, contentType, extensionAttributes);
+            }
+            else
+            {
+                throw new ArgumentException("HTTP message does not represent a CloudEvents batch.", nameof(httpListenerRequest));
+            }
+        }
+
         private static bool HasCloudEventsContentType(HttpListenerRequest request) =>
             MimeUtilities.IsCloudEventsContentType(request.ContentType);
 

test/CloudNative.CloudEvents.UnitTests/Http/HttpClientExtensionsTest.cs

@@ -438,7 +438,7 @@ namespace CloudNative.CloudEvents.Http.UnitTests
             }
         }
 
-        private static HttpRequestMessage CreateRequestMessage(ReadOnlyMemory<byte> content, ContentType contentType) =>
+        internal static HttpRequestMessage CreateRequestMessage(ReadOnlyMemory<byte> content, ContentType contentType) =>
             new HttpRequestMessage
             {
                 Content = new ByteArrayContent(content.ToArray())
@@ -447,7 +447,7 @@ namespace CloudNative.CloudEvents.Http.UnitTests
                 }
             };
 
-        private static HttpResponseMessage CreateResponseMessage(ReadOnlyMemory<byte> content, ContentType contentType) =>
+        internal static HttpResponseMessage CreateResponseMessage(ReadOnlyMemory<byte> content, ContentType contentType) =>
             new HttpResponseMessage
             {
                 Content = new ByteArrayContent(content.ToArray())

test/CloudNative.CloudEvents.UnitTests/Http/HttpListenerExtensionsTest.cs

@@ -141,6 +141,48 @@ namespace CloudNative.CloudEvents.Http.UnitTests
             Assert.Equal(originalCloudEvent.Data, parsedCloudEvent.Data);
         }
 
+        [Fact]
+        public async Task ToCloudEventBatch_Valid()
+        {
+            var batch = CreateSampleBatch();
+
+            var formatter = new JsonEventFormatter();
+            var contentBytes = formatter.EncodeBatchModeMessage(batch, out var contentType);
+
+            AssertBatchesEqual(batch, await GetBatchAsync(context => context.Request.ToCloudEventBatchAsync(formatter, EmptyExtensionArray)));
+            AssertBatchesEqual(batch, await GetBatchAsync(context => context.Request.ToCloudEventBatchAsync(formatter, EmptyExtensionSequence)));
+            AssertBatchesEqual(batch, await GetBatchAsync(context => Task.FromResult(context.Request.ToCloudEventBatch(formatter, EmptyExtensionArray))));
+            AssertBatchesEqual(batch, await GetBatchAsync(context => Task.FromResult(context.Request.ToCloudEventBatch(formatter, EmptyExtensionSequence))));
+
+            Task<IReadOnlyList<CloudEvent>> GetBatchAsync(Func<HttpListenerContext, Task<IReadOnlyList<CloudEvent>>> handler)
+            {
+                var request = HttpClientExtensionsTest.CreateRequestMessage(contentBytes, contentType);
+                request.RequestUri = new Uri(ListenerAddress);
+                return SendRequestAsync(request, handler);
+            }
+        }
+
+        [Fact]
+        public async Task ToCloudEventBatchAsync_Invalid()
+        {
+            // Most likely accident: calling ToCloudEventBatchAsync with a single event in structured mode.
+            var cloudEvent = new CloudEvent().PopulateRequiredAttributes();
+            var formatter = new JsonEventFormatter();
+            var contentBytes = formatter.EncodeStructuredModeMessage(cloudEvent, out var contentType);
+
+            await ExpectFailure(context => context.Request.ToCloudEventBatchAsync(formatter, EmptyExtensionArray));
+            await ExpectFailure(context => context.Request.ToCloudEventBatchAsync(formatter, EmptyExtensionSequence));
+            await ExpectFailure(context => Task.FromResult(context.Request.ToCloudEventBatch(formatter, EmptyExtensionArray)));
+            await ExpectFailure(context => Task.FromResult(context.Request.ToCloudEventBatch(formatter, EmptyExtensionSequence)));
+
+            async Task ExpectFailure(Func<HttpListenerContext, Task<IReadOnlyList<CloudEvent>>> handler)
+            {
+                var request = HttpClientExtensionsTest.CreateRequestMessage(contentBytes, contentType);
+                request.RequestUri = new Uri(ListenerAddress);
+                await SendRequestAsync(request, context => Assert.ThrowsAsync<ArgumentException>(() => handler(context)));
+            }
+        }
+
         [Fact]
         public async Task CopyToHttpListenerResponseAsync_BinaryMode()
         {