SerializationContext and IHeader (#784)

* typed header values

* adding to the integration test

* serde interface changes

* minor tweaks

* review changes

* SerializationContext.Empty

Author: mhowlett

examples/Producer/Program.cs

@@ -92,7 +92,7 @@ public static async Task Main(string[] args)
 
                     try
                     {
-                        // Notes: Awaiting the asynchronous produce request below prevents flow of execution
+                        // Note: Awaiting the asynchronous produce request below prevents flow of execution
                         // from proceeding until the acknowledgement from the broker is received (at the 
                         // expense of low throughput).
                         var deliveryReport = await producer.ProduceAsync(

examples/Protobuf/Program.cs

@@ -35,7 +35,7 @@ namespace Confluent.Kafka.Examples.Protobuf
     /// </summary>
     public class ProtobufSerializer<T> : ISerializer<T> where T : IMessage<T>, new()
     {
-        public byte[] Serialize(T data, bool isKey, MessageMetadata messageMetadata, TopicPartition destination)
+        public byte[] Serialize(T data, SerializationContext context)
             => data.ToByteArray();
     }
 
@@ -51,7 +51,7 @@ public ProtobufDeserializer()
             parser = new MessageParser<T>(() => new T());
         }
 
-        public T Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+        public T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             => parser.ParseFrom(data.ToArray());
     }
 

src/Confluent.Kafka/Consumer.cs

@@ -905,7 +905,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                                 headerValue = new byte[(int)sizep];
                                 Marshal.Copy(valuep, headerValue, 0, (int)sizep);
                             }
-                            headers.Add(new Header(headerName, headerValue));
+                            headers.Add(headerName, headerValue);
                         }
                     }
                 }
@@ -938,9 +938,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                                 ? ReadOnlySpan<byte>.Empty
                                 : new ReadOnlySpan<byte>(msg.key.ToPointer(), (int)msg.key_len),
                             msg.key == IntPtr.Zero,
-                            true,
-                            new MessageMetadata { Timestamp = timestamp, Headers = headers },
-                            new TopicPartition(topic, msg.partition));
+                            new SerializationContext(MessageComponentType.Key, topic));
                     }
                 }
                 catch (Exception exception)
@@ -972,9 +970,7 @@ private ConsumeResult<K, V> ConsumeImpl<K,V>(
                                 ? ReadOnlySpan<byte>.Empty
                                 : new ReadOnlySpan<byte>(msg.val.ToPointer(), (int)msg.len),
                             msg.val == IntPtr.Zero,
-                            false,
-                            new MessageMetadata { Timestamp = timestamp, Headers = headers },
-                            new TopicPartition(topic, msg.partition));
+                            new SerializationContext(MessageComponentType.Value, topic));
                     }
                 }
                 catch (Exception exception)
@@ -1032,15 +1028,15 @@ private ConsumeResult<TKey, TValue> ConsumeViaBytes(int millisecondsTimeout)
             }
 
             TKey key = keyDeserializer != null
-                ? keyDeserializer.Deserialize(rawResult.Key, rawResult.Key == null, true, rawResult.Message, rawResult.TopicPartition)
-                : asyncKeyDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Key), rawResult.Key == null, true, rawResult.Message, rawResult.TopicPartition)
+                ? keyDeserializer.Deserialize(rawResult.Key, rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic))
+                : asyncKeyDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Key), rawResult.Key == null, new SerializationContext(MessageComponentType.Key, rawResult.Topic))
                     .ConfigureAwait(continueOnCapturedContext: false)
                     .GetAwaiter()
                     .GetResult();
 
             TValue val = valueDeserializer != null
-                ? valueDeserializer.Deserialize(rawResult.Value, rawResult.Value == null, false, rawResult.Message, rawResult.TopicPartition)
-                : asyncValueDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Value), rawResult == null, false, rawResult.Message, rawResult.TopicPartition)
+                ? valueDeserializer.Deserialize(rawResult.Value, rawResult.Value == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic))
+                : asyncValueDeserializer.DeserializeAsync(new ReadOnlyMemory<byte>(rawResult.Value), rawResult == null, new SerializationContext(MessageComponentType.Value, rawResult.Topic))
                     .ConfigureAwait(continueOnCapturedContext: false)
                     .GetAwaiter()
                     .GetResult();

src/Confluent.Kafka/Deserializers.cs

@@ -32,7 +32,7 @@ public static class Deserializers
 
         private class Utf8Deserializer : IDeserializer<string>
         {
-            public string Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public string Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull)
                 {
@@ -65,7 +65,7 @@ public string Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Mess
 
         private class NullDeserializer : IDeserializer<Null>
         {
-            public Null Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public Null Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (!isNull)
                 {
@@ -83,7 +83,7 @@ public Null Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Messag
 
         private class IgnoreDeserializer : IDeserializer<Ignore>
         {
-            public Ignore Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public Ignore Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
                 => null;
         }
 
@@ -94,7 +94,7 @@ public Ignore Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Mess
 
         private class Int64Deserializer : IDeserializer<long>
         {
-            public long Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public long Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull)
                 {
@@ -126,7 +126,7 @@ public long Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Messag
 
         private class Int32Deserializer : IDeserializer<int>
         {
-            public int Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public int Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull)
                 {
@@ -154,7 +154,7 @@ public int Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Message
 
         private class SingleDeserializer : IDeserializer<float>
         {
-            public float Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public float Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull)
                 {
@@ -205,7 +205,7 @@ public float Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Messa
 
         private class DoubleDeserializer : IDeserializer<double>
         {
-            public double Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public double Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull)
                 {
@@ -263,7 +263,7 @@ public double Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, Mess
 
         private class ByteArrayDeserializer : IDeserializer<byte[]>
         {
-            public byte[] Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source)
+            public byte[] Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context)
             {
                 if (isNull) { return null; }
                 return data.ToArray();

src/Confluent.Kafka/Header.cs

@@ -25,18 +25,23 @@ namespace Confluent.Kafka
     /// <remarks>
     ///     Message headers are supported by v0.11 brokers and above.
     /// </remarks>
-    public class Header
+    public class Header : IHeader
     {
+        private byte[] val;
+
         /// <summary>
         ///     The header key.
         /// </summary>
         public string Key { get; private set; }
 
         /// <summary>
-        ///     The header value.
+        ///     Get the serialized header value data.
         /// </summary>
-        public byte[] Value { get; private set; }
-
+        public byte[] GetValueBytes()
+        {
+            return val;
+        }
+        
         /// <summary>
         ///     Create a new Header instance.
         /// </summary>
@@ -54,7 +59,7 @@ public Header(string key, byte[] value)
             }
 
             Key = key;
-            Value = value;
+            val = value;
         }
     }
 }

src/Confluent.Kafka/Headers.cs

@@ -27,9 +27,9 @@ namespace Confluent.Kafka
     /// <remarks>
     ///     Message headers are supported by v0.11 brokers and above.
     /// </remarks>
-    public class Headers : IEnumerable<Header>
+    public class Headers : IEnumerable<IHeader>
     {
-        private readonly List<Header> headers = new List<Header>();
+        private readonly List<IHeader> headers = new List<IHeader>();
 
         /// <summary>
         ///     Append a new header to the collection.
@@ -75,9 +75,9 @@ public void Add(Header header)
         /// <exception cref="KeyNotFoundException">
         ///     The key <paramref name="key" /> was not present in the collection.
         /// </exception>
-        public byte[] GetLast(string key)
+        public byte[] GetLastBytes(string key)
         {
-            if (TryGetLast(key, out byte[] result))
+            if (TryGetLastBytes(key, out byte[] result))
             {
                 return result;
             }
@@ -100,21 +100,22 @@ public byte[] GetLast(string key)
         ///     true if the a value with the specified key was present in 
         ///     the collection, false otherwise.
         /// </returns>
-        public bool TryGetLast(string key, out byte[] lastHeader)
+        public bool TryGetLastBytes(string key, out byte[] lastHeader)
         {
             for (int i=headers.Count-1; i>=0; --i)
             {
                 if (headers[i].Key == key)
                 {
-                    lastHeader = headers[i].Value;
+                    lastHeader = headers[i].GetValueBytes();
                     return true;
                 }
             }
 
-            lastHeader = null;
+            lastHeader = default(byte[]);
             return false;
         }
 
+
         /// <summary>
         ///     Removes all headers for the given key.
         /// </summary>
@@ -124,7 +125,7 @@ public bool TryGetLast(string key, out byte[] lastHeader)
         public void Remove(string key)
             => headers.RemoveAll(a => a.Key == key);
 
-        internal class HeadersEnumerator : IEnumerator<Header>
+        internal class HeadersEnumerator : IEnumerator<IHeader>
         {
             private Headers headers;
 
@@ -136,10 +137,10 @@ public HeadersEnumerator(Headers headers)
             }
 
             public object Current 
-                => ((IEnumerator<Header>)this).Current;
+                => ((IEnumerator<IHeader>)this).Current;
 
-            Header IEnumerator<Header>.Current
-                => new Header(headers.headers[location].Key, headers.headers[location].Value);
+            IHeader IEnumerator<IHeader>.Current
+                => headers.headers[location];
 
             public void Dispose() {}
 
@@ -166,7 +167,7 @@ public void Reset()
         /// <returns>
         ///     An enumerator object that can be used to iterate through the headers collection.
         /// </returns>
-        public IEnumerator<Header> GetEnumerator()
+        public IEnumerator<IHeader> GetEnumerator()
             => new HeadersEnumerator(this);
 
         /// <summary>
@@ -184,7 +185,7 @@ IEnumerator IEnumerable.GetEnumerator()
         /// <param key="index">
         ///     The zero-based index of the element to get.
         /// </param>
-        public Header this[int index]
+        public IHeader this[int index]
             => headers[index];
 
         /// <summary>

src/Confluent.Kafka/IAsyncDeserializer.cs

@@ -34,21 +34,13 @@ public interface IAsyncDeserializer<T>
         /// <param name="isNull">
         ///     True if this is a null value.
         /// </param>
-        /// <param name="messageMetadata">
-        ///     Properties of the message the data is associated with
-        ///     extra to the key or value.
-        /// </param>
-        /// <param name="source">
-        ///     The TopicPartition from which the message was consumed.
-        /// </param>
-        /// <param name="isKey">
-        ///     True if deserializing the message key, false if deserializing the
-        ///     message value.
+        /// <param name="context">
+        ///     Context relevant to the deserialize operation.
         /// </param>
         /// <returns>
         ///     A <see cref="System.Threading.Tasks.Task" /> that completes
         ///     with the deserialized value.
         /// </returns>
-        Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source);
+        Task<T> DeserializeAsync(ReadOnlyMemory<byte> data, bool isNull, SerializationContext context);
     }
 }

src/Confluent.Kafka/IAsyncSerializer.cs

@@ -32,22 +32,13 @@ public interface IAsyncSerializer<T>
         /// <param name="data">
         ///     The value to serialize.
         /// </param>
-        /// <param name="messageMetadata">
-        ///     Properties of the message the data is associated with
-        ///     extra to the key or value.
-        /// </param>
-        /// <param name="destination">
-        ///     The TopicPartition to which the message is to be sent
-        ///     (partition may be Partition.Any).
-        /// </param>
-        /// <param name="isKey">
-        ///     True if serializing the message key, false if serializing the
-        ///     message value.
+        /// <param name="context">
+        ///     Context relevant to the serialize operation.
         /// </param>
         /// <returns>
         ///     A <see cref="System.Threading.Tasks.Task" /> that
         ///     completes with the serialized data.
         /// </returns>
-        Task<byte[]> SerializeAsync(T data, bool isKey, MessageMetadata messageMetadata, TopicPartition destination);
+        Task<byte[]> SerializeAsync(T data, SerializationContext context);
     }
 }

src/Confluent.Kafka/IDeserializer.cs

@@ -33,20 +33,12 @@ public interface IDeserializer<T>
         /// <param name="isNull">
         ///     Whether or not the value is null.
         /// </param>
-        /// <param name="messageMetadata">
-        ///     Properties of the message the data is associated with
-        ///     extra to the key or value.
-        /// </param>
-        /// <param name="source">
-        ///     The TopicPartition from which the message was consumed.
-        /// </param>
-        /// <param name="isKey">
-        ///     True if deserializing the message key, false if deserializing the
-        ///     message value.
+        /// <param name="context">
+        ///     Context relevant to the deserialize operation.
         /// </param>
         /// <returns>
         ///     The deserialized value.
         /// </returns>
-        T Deserialize(ReadOnlySpan<byte> data, bool isNull, bool isKey, MessageMetadata messageMetadata, TopicPartition source);
+        T Deserialize(ReadOnlySpan<byte> data, bool isNull, SerializationContext context);
     }
 }

src/Confluent.Kafka/IHeader.cs

@@ -0,0 +1,35 @@
+// Copyright 2019 Confluent Inc.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+//
+// Refer to LICENSE for more information.
+
+
+namespace Confluent.Kafka
+{
+    /// <summary>
+    ///     Defines a Kafka message header.
+    /// </summary>
+    public interface IHeader
+    {
+        /// <summary>
+        ///     The header key.
+        /// </summary>
+        string Key { get; }
+        
+        /// <summary>
+        ///     The serialized header value data.
+        /// </summary>
+        byte[] GetValueBytes();
+    }
+}