Fix case sensitivity issues

- Web hook method must be exactly "OPTIONS"
- Kafka header keys are case sensitive

Fixes #90

Signed-off-by: Jon Skeet <[email protected]>

Author: jskeet

src/CloudNative.CloudEvents.Kafka/KafkaClientExtensions.cs

@@ -23,8 +23,6 @@ public static class KafkaClientExtensions
         private const string KafkaContentTypeAttributeName = "content-type";
         private const string SpecVersionKafkaHeader = KafkaHeaderPrefix + "specversion";
 
-        // TODO: Avoid all the byte[] -> string conversions? If we didn't care about case-sensitivity, we could prepare byte arrays to perform comparisons with.
-
         /// <summary>
         /// Indicates whether this message holds a single CloudEvent.
         /// </summary>
@@ -35,7 +33,7 @@ public static class KafkaClientExtensions
         /// <returns>true, if the request is a CloudEvent</returns>
         public static bool IsCloudEvent(this Message<string, byte[]> message) =>
             GetHeaderValue(message, SpecVersionKafkaHeader) is object ||
-            MimeUtilities.IsCloudEventsContentType(ExtractContentType(message));
+            MimeUtilities.IsCloudEventsContentType(GetHeaderValue(message, KafkaContentTypeAttributeName));
 
         /// <summary>
         /// Converts this Kafka message into a CloudEvent object.
@@ -66,7 +64,7 @@ public static CloudEvent ToCloudEvent(this Message<string, byte[]> message,
                 throw new InvalidOperationException();
             }
 
-            var contentType = ExtractContentType(message);
+            var contentType = GetHeaderValue(message, KafkaContentTypeAttributeName);
 
             CloudEvent cloudEvent;
 
@@ -78,11 +76,10 @@ public static CloudEvent ToCloudEvent(this Message<string, byte[]> message,
             else
             {
                 // Binary mode
-                if (!(GetHeaderValue(message, SpecVersionKafkaHeader) is byte[] versionIdBytes))
+                if (!(GetHeaderValue(message, SpecVersionKafkaHeader) is string versionId))
                 {
                     throw new ArgumentException("Request is not a CloudEvent");
                 }
-                string versionId = Encoding.UTF8.GetString(versionIdBytes);
                 CloudEventsSpecVersion version = CloudEventsSpecVersion.FromVersionId(versionId)
                     ?? throw new ArgumentException($"Unknown CloudEvents spec version '{versionId}'", nameof(message));
 
@@ -115,12 +112,6 @@ public static CloudEvent ToCloudEvent(this Message<string, byte[]> message,
             return Validation.CheckCloudEventArgument(cloudEvent, nameof(message));
         }
 
-        private static string ExtractContentType(Message<string, byte[]> message)
-        {
-            var headerValue = GetHeaderValue(message, KafkaContentTypeAttributeName);
-            return headerValue is null ? null : Encoding.UTF8.GetString(headerValue);
-        }
-
         private static void InitPartitioningKey(Message<string, byte[]> message, CloudEvent cloudEvent)
         {
             if (!string.IsNullOrEmpty(message.Key))
@@ -129,9 +120,13 @@ private static void InitPartitioningKey(Message<string, byte[]> message, CloudEv
             }
         }
 
-        private static byte[] GetHeaderValue(MessageMetadata message, string headerName) =>
-            message.Headers.FirstOrDefault(x => string.Equals(x.Key, headerName, StringComparison.InvariantCultureIgnoreCase))
-                ?.GetValueBytes();
+        /// <summary>
+        /// Returns the last header value with the given name, decoded using UTF-8, or null if there is no such header.
+        /// </summary>
+        private static string GetHeaderValue(MessageMetadata message, string headerName) =>
+            Validation.CheckNotNull(message, nameof(message)).Headers is null
+            ? null
+            : message.Headers.TryGetLastBytes(headerName, out var bytes) ? Encoding.UTF8.GetString(bytes) : null;
 
         /// <summary>
         /// Converts a CloudEvent to a Kafka message.

src/CloudNative.CloudEvents/Http/HttpClientExtensions.cs

@@ -108,7 +108,7 @@ public static bool IsCloudEventBatch(this HttpResponseMessage httpResponseMessag
         /// Indicates whether this HttpListenerRequest is a web hook validation request
         /// </summary>
         public static bool IsWebHookValidationRequest(this HttpRequestMessage httpRequestMessage) =>
-            httpRequestMessage.Method.Method.Equals("options", StringComparison.InvariantCultureIgnoreCase) &&
+            httpRequestMessage.Method.Method == "OPTIONS" &&
             httpRequestMessage.Headers.Contains("WebHook-Request-Origin");
 
         /// <summary>

src/CloudNative.CloudEvents/Http/HttpListenerExtensions.cs

@@ -140,7 +140,7 @@ public static bool IsCloudEvent(this HttpListenerRequest httpListenerRequest) =>
         /// Indicates whether this HttpListenerRequest is a web hook validation request
         /// </summary>
         public static bool IsWebHookValidationRequest(this HttpListenerRequest httpRequestMessage) =>
-            httpRequestMessage.HttpMethod.Equals("options", StringComparison.InvariantCultureIgnoreCase) &&
+            httpRequestMessage.HttpMethod == "OPTIONS" &&
             httpRequestMessage.Headers["WebHook-Request-Origin"] is object;
 
         /// <summary>

test/CloudNative.CloudEvents.UnitTests/Kafka/KafkaTest.cs

@@ -17,6 +17,25 @@ namespace CloudNative.CloudEvents.Kafka.UnitTests
 {
     public class KafkaTest
     {
+        [Theory]
+        [InlineData("content-type", "application/cloudevents", true)]
+        [InlineData("content-type", "APPLICATION/CLOUDEVENTS", true)]
+        [InlineData("CONTENT-TYPE", "application/cloudevents", false)]
+        [InlineData("ce_specversion", "1.0", true)]
+        [InlineData("CE_SPECVERSION", "1.0", false)]
+        public void IsCloudEvent(string headerName, string headerValue, bool expectedResult)
+        {
+            var message = new Message<string, byte[]>
+            {
+                Headers = new Headers { { headerName, Encoding.UTF8.GetBytes(headerValue) } }
+            };
+            Assert.Equal(expectedResult, message.IsCloudEvent());
+        }
+
+        [Fact]
+        public void IsCloudEvent_NoHeaders() =>
+            Assert.False(new Message<string, byte[]>().IsCloudEvent());
+
         [Fact]
         public void KafkaStructuredMessageTest()
         {