merged + updated strongly typed config classes

Author: Matt Howlett

README.md

@@ -90,7 +90,7 @@ class Program
 {
     public static async Task Main(string[] args)
     {
-        var config = new Dictionary<string, object> { { "bootstrap.servers", "localhost:9092" } };
+        var config = new ProducerConfig { BootstrapServers = "localhost:9092" };
 
         // A Producer for sending messages with null keys and UTF-8 encoded values.
         using (var p = new Producer<Null, string>(config))
@@ -109,8 +109,8 @@ class Program
 }
 ```
 
-However, a server round-trip is slow (3ms at a minimum; actual latency depends on many factors).
-In highly concurrent scenarios you will get high overall throughput out of the producer using 
+Note that a server round-trip is slow (3ms at a minimum; actual latency depends on many factors).
+In highly concurrent scenarios you will achieve high overall throughput out of the producer using 
 the above approach, but there will be a delay on each `await` call. In stream processing 
 applications, where you would like to process many messages in rapid succession, you would typically
 make use the `BeginProduce` method instead:
@@ -126,7 +126,7 @@ class Program
 {
     public static void Main(string[] args)
     {
-        var conf = new Dictionary<string, object> { { "bootstrap.servers", "localhost:9092" } };
+        var conf = new ProducerConfig { BootstrapServers = "localhost:9092" };
 
         Action<DeliveryReportResult<Null, string>> handler = r => 
             Console.WriteLine(!r.Error.IsError
@@ -161,11 +161,11 @@ class Program
 {
     public static void Main(string[] args)
     {
-        var conf = new Dictionary<string, object> 
+        var conf = new ConsumerConfig
         { 
-            { "group.id", "test-consumer-group" },
-            { "bootstrap.servers", "localhost:9092" },
-            { "auto.offset.reset", "earliest" }
+            GroupId = "test-consumer-group",
+            BootstrapServers = "localhost:9092",
+            AutoOffsetReset = AutoOffsetResetType.Earliest
         };
 
         using (var c = new Consumer<Ignore, string>(conf))

examples/AvroBlogExamples/Program.cs

@@ -38,7 +38,7 @@ static void ProduceGeneric(string bootstrapServers, string schemaRegistryUrl)
             var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
 
             using (var serdeProvider = new AvroSerdeProvider(new AvroSerdeProviderConfig { SchemaRegistryUrl = schemaRegistryUrl }))
-            using (var producer = new Producer<Null, GenericRecord>(producerConfig, null, serdeProvider.CreateValueSerializer<GenericRecord>()))
+            using (var producer = new Producer<Null, GenericRecord>(producerConfig, null, serdeProvider.GetSerializerGenerator<GenericRecord>()))
             {
                 var logLevelSchema = (EnumSchema)Schema.Parse(
                     File.ReadAllText("LogLevel.asvc"));
@@ -68,7 +68,7 @@ static void ProduceSpecific(string bootstrapServers, string schemaRegistryUrl)
             var producerConfig = new ProducerConfig { BootstrapServers = bootstrapServers };
 
             using (var serdeProvider = new AvroSerdeProvider(new AvroSerdeProviderConfig { SchemaRegistryUrl = schemaRegistryUrl }))
-            using (var producer = new Producer<Null, MessageTypes.LogMessage>(producerConfig, null, serdeProvider.CreateValueSerializer<MessageTypes.LogMessage>()))
+            using (var producer = new Producer<Null, MessageTypes.LogMessage>(producerConfig, null, serdeProvider.GetSerializerGenerator<MessageTypes.LogMessage>()))
             {
                 producer.ProduceAsync("log-messages", 
                     new Message<Null, MessageTypes.LogMessage> 
@@ -101,7 +101,7 @@ static void ConsumeSpecific(string bootstrapServers, string schemaRegistryUrl)
             };
 
             using (var serdeProvider = new AvroSerdeProvider(new AvroSerdeProviderConfig { SchemaRegistryUrl = schemaRegistryUrl }))
-            using (var consumer = new Consumer<Null, MessageTypes.LogMessage>(consumerConfig, null, serdeProvider.CreateDeserializer<MessageTypes.LogMessage>()))
+            using (var consumer = new Consumer<Null, MessageTypes.LogMessage>(consumerConfig, null, serdeProvider.GetDeserializerGenerator<MessageTypes.LogMessage>()))
             {
                 consumer.Subscribe("log-messages");
 

examples/AvroGeneric/Program.cs

@@ -78,7 +78,7 @@ static void Main(string[] args)
             var consumeTask = Task.Run(() =>
             {
                 using (var serdeProvider = new AvroSerdeProvider(avroConfig))
-                using (var consumer = new Consumer<string, GenericRecord>(consumerConfig, serdeProvider.CreateDeserializer<string>(), serdeProvider.CreateDeserializer<GenericRecord>()))
+                using (var consumer = new Consumer<string, GenericRecord>(consumerConfig, serdeProvider.GetDeserializerGenerator<string>(), serdeProvider.GetDeserializerGenerator<GenericRecord>()))
                 {
                     consumer.OnError += (_, e)
                         => Console.WriteLine($"Error: {e.Reason}");
@@ -103,7 +103,7 @@ static void Main(string[] args)
             }, cts.Token);            
 
             using (var serdeProvider = new AvroSerdeProvider(avroConfig))
-            using (var producer = new Producer<string, GenericRecord>(producerConfig, serdeProvider.CreateKeySerializer<string>(), serdeProvider.CreateValueSerializer<GenericRecord>()))
+            using (var producer = new Producer<string, GenericRecord>(producerConfig, serdeProvider.GetSerializerGenerator<string>(), serdeProvider.GetSerializerGenerator<GenericRecord>()))
             {
                 Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit.");
 

examples/AvroSpecific/Program.cs

@@ -70,7 +70,7 @@ static void Main(string[] args)
             var consumeTask = Task.Run(() =>
             {
                 using (var serdeProvider = new AvroSerdeProvider(avroConfig))
-                using (var consumer = new Consumer<string, User>(consumerConfig, serdeProvider.CreateDeserializer<string>(), serdeProvider.CreateDeserializer<User>()))
+                using (var consumer = new Consumer<string, User>(consumerConfig, serdeProvider.GetDeserializerGenerator<string>(), serdeProvider.GetDeserializerGenerator<User>()))
                 {
                     consumer.OnError += (_, e)
                         => Console.WriteLine($"Error: {e.Reason}");
@@ -95,7 +95,7 @@ static void Main(string[] args)
             }, cts.Token);
 
             using (var serdeProvider = new AvroSerdeProvider(avroConfig))
-            using (var producer = new Producer<string, User>(producerConfig, serdeProvider.CreateKeySerializer<string>(), serdeProvider.CreateValueSerializer<User>()))
+            using (var producer = new Producer<string, User>(producerConfig, serdeProvider.GetSerializerGenerator<string>(), serdeProvider.GetSerializerGenerator<User>()))
             {
                 Console.WriteLine($"{producer.Name} producing on {topicName}. Enter user names, q to exit.");
 

examples/ConfluentCloud/Program.cs

@@ -18,7 +18,6 @@
 using System.Collections.Generic;
 using System.Text;
 using Confluent.Kafka;
-using Confluent.Kafka.Serialization;
 
 
 namespace ConfluentCloudExample

examples/Consumer/Program.cs

@@ -22,7 +22,6 @@
 using System.Text;
 using System.Threading;
 using System.Threading.Tasks;
-using Confluent.Kafka.Serialization;
 
 
 /// <summary>

examples/MultiProducer/Program.cs

@@ -17,7 +17,6 @@
 using System;
 using System.Text;
 using System.Collections.Generic;
-using Confluent.Kafka.Serialization;
 
 
 namespace Confluent.Kafka.Examples.MultiProducer
@@ -35,7 +34,7 @@ public static void Main(string[] args)
             using (var producer = new Producer<string, string>(config))
             {
                 // create a producer of different type that reuses producer's Handle.
-                var producer2 = new Producer<Null, int>(producer.Handle, null, new IntSerializer());
+                var producer2 = new Producer<Null, int>(producer.Handle, null, Serializers.Int32);
 
                 // write (string, string) data to topic "first-topic", statically type checked.
                 producer.ProduceAsync("first-topic", new Message<string, string> { Key = "my-key-value", Value = "my-value" });

examples/Producer/Program.cs

@@ -22,7 +22,6 @@
 using System.Threading.Tasks;
 using System.Collections.Generic;
 using Confluent.Kafka;
-using Confluent.Kafka.Serialization;
 
 
 namespace Confluent.Kafka.Examples.Producer

src/Confluent.Kafka.Avro/AvroDeserializer.cs

@@ -34,7 +34,7 @@ namespace Confluent.Kafka.Serialization
     ///       bytes 1-4:        Unique global id of the avro schema that was used for encoding (as registered in Confluent Schema Registry), big endian.
     ///       following bytes:  The serialized data.
     /// </remarks>
-    public class AvroDeserializer<T> : IDeserializer<T>
+    public class AvroDeserializer<T>
     {
         private ISchemaRegistryClient schemaRegistryClient;
 
@@ -82,7 +82,7 @@ public T Deserialize(string topic, ReadOnlySpan<byte> data, bool isNull)
         }
 
         /// <summary>
-        ///     Refer to: <see cref="Confluent.Kafka.Serialization.ISerializer{T}.Configure(IEnumerable{KeyValuePair{string, string}}, bool)" />
+        ///     Configure the deserializer.
         /// </summary>
         public IEnumerable<KeyValuePair<string, string>> Configure(IEnumerable<KeyValuePair<string, string>> config, bool isKey)
         {

src/Confluent.Kafka.Avro/AvroSerdeProvider.cs

@@ -14,7 +14,7 @@ namespace Confluent.Kafka.Serialization
     /// </summary>
     public class AvroSerdeProvider : IDisposable
     {
-        AvroSerdeProviderConfig config;
+        IEnumerable<KeyValuePair<string, string>> config;
         ISchemaRegistryClient schemaRegistryClient;
 
         /// <summary>
@@ -23,14 +23,14 @@ public class AvroSerdeProvider : IDisposable
         /// <param name="config">
         ///     Configuration properties.
         /// </param>
-        public AvroSerdeProvider(AvroSerdeProviderConfig config)
+        public AvroSerdeProvider(IEnumerable<KeyValuePair<string, string>> config)
         {
             this.config = config;
             schemaRegistryClient = new CachedSchemaRegistryClient(config);
         }
 
         /// <summary>
-        ///     Create a new avro deserializer. Use this deserializer with 
+        ///     Create a new avro deserializer generator. Use this with 
         ///     GenericRecord, types generated using the avrogen.exe tool or
         ///     one of the following primitive types: int, long, float,
         ///     double, boolean, string, byte[].
@@ -41,46 +41,35 @@ public AvroSerdeProvider(AvroSerdeProviderConfig config)
         ///       bytes 1-4:        Unique global id of the avro schema that was used for encoding (as registered in Confluent Schema Registry), big endian.
         ///       following bytes:  The serialized data.
         /// </remarks>
-        public IDeserializer<T> CreateDeserializer<T>()
+        public DeserializerGenerator<T> GetDeserializerGenerator<T>()
         {
-            var deserializer = new AvroDeserializer<T>(schemaRegistryClient);
-            deserializer.Configure(config, false);
-            return deserializer;
+            return (forKey) => 
+            {
+                var deserializer = new AvroDeserializer<T>(schemaRegistryClient);
+                deserializer.Configure(config, forKey);
+                return (string topic, ReadOnlySpan<byte> data, bool isNull) => deserializer.Deserialize(topic, data, isNull);
+            };
         }
 
         /// <summary>
-        ///     Create a new avro serializer for message keys. Use this serializer
-        ///     with GenericRecord, types generated using the avrogen.exe tool or
+        ///     Create a new avro serializer generator. Use this with
+        ///     GenericRecord, types generated using the avrogen.exe tool or
         ///     one of the following primitive types: int, long, float, double, 
         ///     boolean, string, byte[].
         /// </summary>
-        public ISerializer<T> CreateKeySerializer<T>()
+        public SerializerGenerator<T> GetSerializerGenerator<T>()
         {
-            var serializer = new AvroSerializer<T>(schemaRegistryClient);
-            serializer.Configure(config, true);
-            return serializer;
+            return (forKey) =>
+            {
+                var serializer = new AvroSerializer<T>(schemaRegistryClient);
+                serializer.Configure(config, true);
+                return (string topic, T data) => serializer.Serialize(topic, data);
+            };
         }
 
-        /// <summary>
-        ///     Create a new avro serializer for message values. Use this serializer
-        ///     with GenericRecord, types generated using the avrogen.exe tool or
-        ///     one of the following primitive types: int, long, float, double, 
-        ///     boolean, string, byte[].
-        /// </summary>
-        public ISerializer<T> CreateValueSerializer<T>()
-        {
-            var serializer = new AvroSerializer<T>(schemaRegistryClient);
-            serializer.Configure(config, false);
-            return serializer;
-        }
-
         /// <summary>
         ///     Releases all resources owned by this object.
         /// </summary>
-        public void Dispose()
-        {
-            schemaRegistryClient.Dispose();
-        }
+        public void Dispose() => schemaRegistryClient.Dispose();
     }
 }
-