tweaks

Author: Matt Howlett

doc/docfx.json

@@ -4,7 +4,7 @@
       "src": [
         {
           "files": [
-            "src/Confluent.Kafka/*.cs"
+            "src/**/*.cs"
           ],
           "cwd": ".."
         }

src/Confluent.Kafka/Consumer.cs

@@ -35,7 +35,7 @@ namespace Confluent.Kafka
     ///     consumer <see cref="Confluent.Kafka.Consumer{TKey,TValue}" /> where possible
     ///     (use the byte[] deserializer).
     /// </summary>
-    public class Consumer : IConsumer, IDisposable
+    public class Consumer : IConsumer
     {
         private SafeKafkaHandle kafkaHandle;
 

src/Confluent.Kafka/Consumer_KV.cs

@@ -31,7 +31,7 @@ namespace Confluent.Kafka
     ///     Implements a high-level Apache Kafka consumer (with 
     ///     key and value deserialization).
     /// </summary>
-    public class Consumer<TKey, TValue> : IConsumer<TKey, TValue>, IDisposable
+    public class Consumer<TKey, TValue> : IConsumer<TKey, TValue>
     {
         private readonly Consumer consumer;
 

src/Confluent.Kafka/ErrorCode.cs

@@ -233,6 +233,26 @@ public enum ErrorCode
         /// </summary>
         Local_ValueDeserialization = -159,
 
+        /// <summary>
+        ///     Partial response
+        /// </summary>
+        Local_Partial = -158,
+
+        /// <summary>
+        ///     Modification attempted on read-only object
+        /// </summary>
+        Local_ReadOnly = -157,
+
+        /// <summary>
+        ///     No such entry / item not found
+        /// </summary>
+        Local_NoEnt = -156,
+
+        /// <summary>
+        ///     Read underflow
+        /// </summary>
+        Local_Underflow = -155,
+
 
         /// <summary>
         ///     Unknown broker error

src/Confluent.Kafka/Headers.cs

@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
-//
 // Refer to LICENSE for more information.
 
 using System.Collections.Generic;

src/Confluent.Kafka/IConsumer.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 Confluent Inc., 2015-2016 Andreas Heider
+// Copyright 2018 Confluent Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
-//
 // Refer to LICENSE for more information.
 
 using System;
@@ -35,7 +33,7 @@ namespace Confluent.Kafka
     ///     consumer <see cref="Confluent.Kafka.IConsumer{TKey,TValue}" /> where possible
     ///     (use the byte[] deserializer).
     /// </summary>
-    public interface IConsumer
+    public interface IConsumer : IDisposable
     {
         /// <include file='include_docs_consumer.xml' path='API/Member[@name="OnPartitionsAssigned"]/*' />
         event EventHandler<List<TopicPartition>> OnPartitionsAssigned;

src/Confluent.Kafka/IConsumer_KV.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 Confluent Inc., 2015-2016 Andreas Heider
+// Copyright 2018 Confluent Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
-//
 // Refer to LICENSE for more information.
 
 using System;
@@ -31,7 +29,7 @@ namespace Confluent.Kafka
     ///     Defines a high-level Apache Kafka consumer (with key and 
     ///     value deserialization).
     /// </summary>
-    public interface IConsumer<TKey, TValue>
+    public interface IConsumer<TKey, TValue> : IDisposable
     {
         /// <include file='include_docs_consumer.xml' path='API/Member[@name="KeyDeserializer"]/*' />
         IDeserializer<TKey> KeyDeserializer { get; }

src/Confluent.Kafka/IProducer.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 Confluent Inc., 2015-2016 Andreas Heider
+// Copyright 2018 Confluent Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
-//
 // Refer to LICENSE for more information.
 
 using System;
@@ -31,7 +29,8 @@
 namespace Confluent.Kafka
 {
     /// <summary>
-    ///     Defines a high-level Apache Kafka producer (without serialization).
+    ///     Defines a high-level Apache Kafka producer client (without 
+    ///     serialization capability).
     /// </summary>
     public interface IProducer : IDisposable
     {

src/Confluent.Kafka/IProducer_KV.cs

@@ -1,4 +1,4 @@
-// Copyright 2016-2017 Confluent Inc., 2015-2016 Andreas Heider
+// Copyright 2018 Confluent Inc.
 //
 // Licensed under the Apache License, Version 2.0 (the "License");
 // you may not use this file except in compliance with the License.
@@ -12,8 +12,6 @@
 // See the License for the specific language governing permissions and
 // limitations under the License.
 //
-// Derived from: rdkafka-dotnet, licensed under the 2-clause BSD License.
-//
 // Refer to LICENSE for more information.
 
 using System;
@@ -31,7 +29,7 @@
 namespace Confluent.Kafka
 {
     /// <summary>
-    ///     Defines a high-level Apache Kafka producer with key
+    ///     Defines a high-level Apache Kafka producer client that provides key
     ///     and value serialization.
     /// </summary>
     public interface IProducer<TKey, TValue> : ISerializingProducer<TKey, TValue>, IDisposable

src/Confluent.Kafka/ISerializingProducer.cs

@@ -23,7 +23,8 @@
 namespace Confluent.Kafka
 {
     /// <summary>
-    ///     Describes the minimum functionality provided by a high level (serializing) Kafka producer.
+    ///     A focused interface for producing messages to Kafka with key and 
+    ///     value serialization (excludes general client functionality).
     /// </summary>
     public interface ISerializingProducer<TKey, TValue>
     {
@@ -33,9 +34,6 @@ public interface ISerializingProducer<TKey, TValue>
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ValueSerializer"]/*' />
         ISerializer<TValue> ValueSerializer { get; }
 
-        /// <include file='include_docs_client.xml' path='API/Member[@name="Name"]/*' />
-        string Name { get; }
-
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Message"]/*' />
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Common"]/*' />
         Task<Message<TKey, TValue>> ProduceAsync(Message<TKey, TValue> message);

src/Confluent.Kafka/Ignore.cs

@@ -20,9 +20,8 @@
 namespace Confluent.Kafka
 {
     /// <summary>
-    ///     A type for use in conjunction with <see cref="Confluent.Kafka.Serialization.IgnoreDeserializer" />
-    ///     that enables message keys or values to be read as null, regardless
-    ///     of their value.
+    ///     A type for use in conjunction with IgnoreDeserializer that enables
+    ///     message keys or values to be read as null, regardless of their value.
     /// </summary>
     public sealed class Ignore
     {

src/Confluent.Kafka/Null.cs

@@ -14,15 +14,13 @@
 //
 // Refer to LICENSE for more information.
 
-using Confluent.Kafka.Serialization;
-
 
 namespace Confluent.Kafka
 {
     /// <summary>
-    ///     A type for use in conjunction with <see cref="Confluent.Kafka.Serialization.NullSerializer" />
-    ///     and <see cref="Confluent.Kafka.Serialization.NullDeserializer" /> that enables null key or 
-    ///     values to be enforced when producing or consuming messages.
+    ///     A type for use in conjunction with NullSerializer and NullDeserializer
+    ///     that enables null key or values to be enforced when producing or 
+    ///     consuming messages.
     /// </summary>
     public sealed class Null
     {

src/Confluent.Kafka/Partition.cs

@@ -34,8 +34,7 @@ public struct Partition
         /// <summary>
         ///     A special value that refers to an unspecified / unknown partition.
         /// </summary>
-        /// <returns></returns>
-        public static Partition NotSpecified { get { return new Partition(RD_KAFKA_PARTITION_UA); } }
+        public static Partition Any { get { return new Partition(RD_KAFKA_PARTITION_UA); } }
 
         /// <summary>
         ///     Initializes a new instance of the Partition structure.

src/Confluent.Kafka/Producer.cs

@@ -150,6 +150,24 @@ private static void DeliveryReportCallbackImpl(IntPtr rk, IntPtr rkmessage, IntP
         {
             var msg = Util.Marshal.PtrToStructureUnsafe<rd_kafka_message>(rkmessage);
 
+            Headers headers = new Headers();
+            LibRdKafka.message_headers(rkmessage, out IntPtr hdrsPtr);
+            if (hdrsPtr != IntPtr.Zero)
+            {
+                for (var i=0; ; ++i)
+                {
+                    var err = LibRdKafka.header_get_all(hdrsPtr, (IntPtr)i, out IntPtr namep, out IntPtr valuep, out IntPtr sizep);
+                    if (err != ErrorCode.NoError)
+                    {
+                        break;
+                    }
+                    var headerName = Util.Marshal.PtrToStringUTF8(namep);
+                    var headerValue = new byte[(int)sizep];
+                    Marshal.Copy(valuep, headerValue, 0, (int)sizep);
+                    headers.Add(new KeyValuePair<string, byte[]>(headerName, headerValue));
+                }
+            }
+
             // the msg._private property has dual purpose. Here, it is an opaque pointer set
             // by Topic.Produce to be an IDeliveryHandler. When Consuming, it's for internal
             // use (hence the name).
@@ -193,7 +211,7 @@ private static void DeliveryReportCallbackImpl(IntPtr rk, IntPtr rkmessage, IntP
                     key,
                     val,
                     new Timestamp(timestamp, (TimestampType)timestampType),
-                    null,
+                    headers,
                     msg.err
                 )
             );
@@ -321,22 +339,22 @@ public Producer(IEnumerable<KeyValuePair<string, object>> config, bool manualPol
                     prop.Key != EnableBackgroundPollPropertyName &&
                     prop.Key != EnableDeliveryReportsPropertyName);
 
-            var enableBackgroundPollStr = (string)config.FirstOrDefault(prop => prop.Key == EnableBackgroundPollPropertyName).Value;
-            if (enableBackgroundPollStr != null)
+            var enableBackgroundPollObj = config.FirstOrDefault(prop => prop.Key == EnableBackgroundPollPropertyName).Value;
+            if (enableBackgroundPollObj != null)
             {
-                this.manualPoll = !bool.Parse(enableBackgroundPollStr);
+                this.manualPoll = !bool.Parse(enableBackgroundPollObj.ToString());
             }
 
-            var enableDeliveryReportsStr = (string)config.FirstOrDefault(prop => prop.Key == EnableDeliveryReportsPropertyName).Value;
-            if (enableDeliveryReportsStr != null)
+            var enableDeliveryReportsObj = config.FirstOrDefault(prop => prop.Key == EnableDeliveryReportsPropertyName).Value;
+            if (enableDeliveryReportsObj != null)
             {
-                this.disableDeliveryReports = !bool.Parse(enableDeliveryReportsStr);
+                this.disableDeliveryReports = !bool.Parse(enableDeliveryReportsObj.ToString());
             }
 
-            var blockIfQueueFullStr = (string)config.FirstOrDefault(prop => prop.Key == BlockIfQueueFullPropertyName).Value;
-            if (blockIfQueueFullStr != null)
+            var blockIfQueueFullObj = config.FirstOrDefault(prop => prop.Key == BlockIfQueueFullPropertyName).Value;
+            if (blockIfQueueFullObj != null)
             {
-                this.blockIfQueueFullPropertyValue = bool.Parse(blockIfQueueFullStr);
+                this.blockIfQueueFullPropertyValue = bool.Parse(blockIfQueueFullObj.ToString());
             }
 
             // Note: Setting default topic configuration properties via default.topic.config is depreciated 
@@ -404,8 +422,9 @@ public Producer(IEnumerable<KeyValuePair<string, object>> config, bool manualPol
         ///     librdkafka configuration parameters (refer to https://github.com/edenhill/librdkafka/blob/master/CONFIGURATION.md).
         /// </param>
         public Producer(IEnumerable<KeyValuePair<string, object>> config)
+#pragma warning disable CS0618
             : this(config, false, false) {}
-
+#pragma warning restore CS0618
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="Poll_int"]/*' />
         public int Poll(int millisecondsTimeout)
@@ -454,7 +473,7 @@ public Task<Message> ProduceAsync(Message message)
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_string_TKey_TValue"]/*' />
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Common"]/*' />
         public Task<Message> ProduceAsync(string topic, byte[] key, byte[] val)
-            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.NotSpecified, null, this.blockIfQueueFullPropertyValue);
+            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.Any, null, this.blockIfQueueFullPropertyValue);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_string_Partition_TKey_TValue_Timestamp_IEnumerable"]/*' />
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Common"]/*' />
@@ -492,7 +511,7 @@ public void Produce(Message message, IDeliveryHandler deliveryHandler)
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_string_TKey_TValue"]/*' />
         /// <include file='include_docs_producer.xml' path='API/Member[@name="Produce_IDeliveryHandler"]/*' />
         public void Produce(string topic, byte[] key, byte[] val, IDeliveryHandler deliveryHandler)
-            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.NotSpecified, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
+            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.Any, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_string_Partition_TKey_TValue_Timestamp_IEnumerable"]/*' />
         /// <include file='include_docs_producer.xml' path='API/Member[@name="Produce_IDeliveryHandler"]/*' />
@@ -520,7 +539,7 @@ IDeliveryHandler deliveryHandler
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("The Producer API has been revised and this overload of ProduceAsync has been depreciated. Please use another variant of ProduceAsync.")]
         public Task<Message> ProduceAsync(string topic, byte[] key, int keyOffset, int keyLength, byte[] val, int valOffset, int valLength)
-            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.NotSpecified, null, this.blockIfQueueFullPropertyValue);
+            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.Any, null, this.blockIfQueueFullPropertyValue);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("The Producer API has been revised and this overload of ProduceAsync has been depreciated. Please use another variant of ProduceAsync.")]
@@ -535,17 +554,17 @@ public Task<Message> ProduceAsync(string topic, byte[] key, int keyOffset, int k
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("Variants of ProduceAsync that include a blockIfQueueFull parameter are depreciated - use the dotnet.producer.block.if.queue.full configuration property instead.")]
         public Task<Message> ProduceAsync(string topic, byte[] key, int keyOffset, int keyLength, byte[] val, int valOffset, int valLength, bool blockIfQueueFull)
-            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.NotSpecified, null, blockIfQueueFull);
+            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.Any, null, blockIfQueueFull);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("Variants of ProduceAsync that include a IDeliveryHandler parameter are depreciated - use a variant of Produce instead.")]
         public void ProduceAsync(string topic, byte[] key, byte[] val, IDeliveryHandler deliveryHandler)
-            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.NotSpecified, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
+            => ProduceImpl(topic, val, 0, val?.Length ?? 0, key, 0, key?.Length ?? 0, Timestamp.Default, Partition.Any, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("Variants of ProduceAsync that include a IDeliveryHandler parameter are depreciated - use a variant of Produce instead.")]
         public void ProduceAsync(string topic, byte[] key, int keyOffset, int keyLength, byte[] val, int valOffset, int valLength, IDeliveryHandler deliveryHandler)
-            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.NotSpecified, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
+            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.Any, null, this.blockIfQueueFullPropertyValue, deliveryHandler);
 
         /// <include file='include_docs_producer.xml' path='API/Member[@name="ProduceAsync_Obsolete"]/*' />
         [Obsolete("Variants of ProduceAsync that include a IDeliveryHandler parameter are depreciated - use a variant of Produce instead.")]
@@ -564,7 +583,7 @@ public void ProduceAsync(string topic, byte[] key, int keyOffset, int keyLength,
             "Variants of ProduceAsync that include a IDeliveryHandler parameter are depreciated - use a variant of Produce instead. " +
             "Variants of ProduceAsync that include a blockIfQueueFull parameter are depreciated - use the dotnet.producer.block.if.queue.full configuration property instead.")]
         public void ProduceAsync(string topic, byte[] key, int keyOffset, int keyLength, byte[] val, int valOffset, int valLength, bool blockIfQueueFull, IDeliveryHandler deliveryHandler)
-            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.NotSpecified, null, blockIfQueueFull, deliveryHandler);
+            => ProduceImpl(topic, val, valOffset, valLength, key, keyOffset, keyLength, Timestamp.Default, Partition.Any, null, blockIfQueueFull, deliveryHandler);
 
 #endregion