Add a web example (#1440)

* WIP: Web example

* Changes following review feedback

* Fix random gen

* efficient -> inefficient

Author: mhowlett

README.md

@@ -200,6 +200,13 @@ class Program
 }
 ```
 
+### IHostedService and Web Application Integration
+
+The [Web](https://github.com/confluentinc/confluent-kafka-dotnet/tree/master/examples/Web) example demonstrates how to integrate
+Apache Kafka with a web application, including how to implement `IHostedService` to realize a long running consumer poll loop, how to
+register a producer as a singleton service, and how to bind configuration from an injected `IConfiguration` instance.
+
+
 ### Schema Registry Integration
 
 The three "Serdes" packages provide serializers and deserializers for Avro, Protobuf and JSON with [Confluent Schema Registry](https://docs.confluent.io/current/schema-registry/docs/index.html) integration. The `Confluent.SchemaRegistry` nuget package provides a client for interfacing with

examples/Web/Controllers/HomeController.cs

@@ -0,0 +1,81 @@
+// Copyright 2020 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.
+
+using System;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+using Confluent.Kafka;
+
+
+namespace Web.Controllers
+{
+    public class HomeController : Controller
+    {
+        private string topic;
+        private readonly KafkaDependentProducer<Null, string> producer;
+        private readonly ILogger logger;
+
+        public HomeController(KafkaDependentProducer<Null, string> producer, IConfiguration config, ILogger<HomeController> logger)
+        {
+            // In a real-world application, you might be using Kafka as a system of record and may wish
+            // to update application state in your request handlers. Or you may wish to write some
+            // analytics data related to your handler logic. In this example, we aren't doing anything
+            // interesting, so we just write frivolous messages to demonstrate how to set things up.
+            this.topic = config.GetValue<string>("Kafka:FrivolousTopic");
+            this.producer = producer;
+            this.logger = logger;
+        }
+
+        public async Task<IActionResult> Index()
+        {
+            // Simulate a complex request handler by delaying a random amount of time.
+            await Task.Delay((int)(new Random((int)DateTime.Now.Ticks).NextDouble()*100));
+
+            // Important note: DO NOT create a new producer instance every time you
+            // need to produce a message (this is a common pattern with relational database
+            // drivers, but it is extremely inefficient here). Instead, use a long-lived
+            // singleton instance, as per this example.
+
+            // Do not delay completion of the page request on the result of produce call.
+            // Any errors are handled out of band in deliveryReportHandler.
+            this.producer.Produce(topic, new Message<Null, string> { Value = "Frivolous message (index)" }, deliveryReportHandler);
+            return View();
+        }
+
+        public async Task<IActionResult> Page1()
+        {
+            await Task.Delay((int)(new Random((int)DateTime.Now.Ticks).NextDouble()*100));
+
+            // Delay completion of the page request on the result of the produce call.
+            // An exception will be thrown in the case of an error.
+            await this.producer.ProduceAsync(topic, new Message<Null, string> { Value = "Frivolous message #1" });
+            return View();
+        }
+
+        private void deliveryReportHandler(DeliveryReport<Null, string> deliveryReport)
+        {
+            if (deliveryReport.Status == PersistenceStatus.NotPersisted)
+            {
+                // It is common to write application logs to Kafka (note: this project does not provide
+                // an example logger implementation that does this). Such an implementation should
+                // ideally fall back to logging messages locally in the case of delivery problems.
+                this.logger.Log(LogLevel.Warning, $"Message delivery failed: {deliveryReport.Message.Value}");
+            }
+        }
+    }
+}

examples/Web/KafkaClientHandle.cs

@@ -0,0 +1,57 @@
+// Copyright 2020 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.
+
+using Confluent.Kafka;
+using Microsoft.Extensions.Configuration;
+
+
+namespace Web
+{
+    /// <summary>
+    ///     Wraps a Confluent.Kafka.IProducer instance, and allows for basic
+    ///     configuration of this via IConfiguration.
+    ///    
+    ///     KafkaClientHandle does not provide any way for messages to be produced
+    ///     directly. Instead, it is a dependency of KafkaDependentProducer. You
+    ///     can create more than one instance of KafkaDependentProducer (with
+    ///     possibly differing K and V generic types) that leverages the same
+    ///     underlying producer instance exposed by the Handle property of this
+    ///     class. This is more efficient than creating separate
+    ///     Confluent.Kafka.IProducer instances for each Message type you wish to
+    ///     produce.
+    /// </summary>
+    public class KafkaClientHandle
+    {
+        IProducer<byte[], byte[]> kafkaProducer;
+
+        public KafkaClientHandle(IConfiguration config)
+        {
+            var conf = new ProducerConfig();
+            config.GetSection("Kafka:ProducerSettings").Bind(conf);
+            this.kafkaProducer = new ProducerBuilder<byte[], byte[]>(conf).Build();
+        }
+
+        public Handle Handle { get => this.kafkaProducer.Handle; }
+
+        public void Dispose()
+        {
+            // Block until all outstanding produce requests have completed (with or
+            // without error).
+            kafkaProducer.Flush();
+            kafkaProducer.Dispose();
+        }
+    }
+}

examples/Web/KafkaDependentProducer.cs

@@ -0,0 +1,57 @@
+// Copyright 2020 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.
+
+using Confluent.Kafka;
+using System;
+using System.Threading.Tasks;
+
+
+namespace Web
+{
+    /// <summary>
+    ///     Leverages the injected KafkaClientHandle instance to allow
+    ///     Confluent.Kafka.Message{K,V}s to be produced to Kafka.
+    /// </summary>
+    public class KafkaDependentProducer<K,V>
+    {
+        IProducer<K, V> kafkaHandle;
+
+        public KafkaDependentProducer(KafkaClientHandle handle)
+        {
+            kafkaHandle = new DependentProducerBuilder<K, V>(handle.Handle).Build();
+        }
+
+        /// <summary>
+        ///     Asychronously produce a message and expose delivery information
+        ///     via the returned Task. Use this method of producing if you would
+        ///     like to await the result before flow of execution continues.
+        /// <summary>
+        public Task ProduceAsync(string topic, Message<K, V> message)
+            => this.kafkaHandle.ProduceAsync(topic, message);
+
+        /// <summary>
+        ///     Asynchronously produce a message and expose delivery information
+        ///     via the provided callback function. Use this method of producing
+        ///     if you would like flow of execution to continue immediately, and
+        ///     handle delivery information out-of-band.
+        /// </summary>
+        public void Produce(string topic, Message<K, V> message, Action<DeliveryReport<K, V>> deliveryHandler = null)
+            => this.kafkaHandle.Produce(topic, message, deliveryHandler);
+
+        public void Flush(TimeSpan timeout)
+            => this.kafkaHandle.Flush(timeout);
+    }
+}

examples/Web/Program.cs

@@ -0,0 +1,36 @@
+// Copyright 2020 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.
+
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Hosting;
+
+namespace Web
+{
+    public class Program
+    {
+        public static void Main(string[] args)
+        {
+            CreateHostBuilder(args).Build().Run();
+        }
+
+        public static IHostBuilder CreateHostBuilder(string[] args) =>
+            Host.CreateDefaultBuilder(args)
+                .ConfigureWebHostDefaults(webBuilder =>
+                {
+                    webBuilder.UseStartup<Startup>();
+                });
+    }
+}

examples/Web/README.md

@@ -0,0 +1,41 @@
+## Web Example
+
+This project demonstrates how to integrate Apache Kafka with a .NET web application. It provides
+examples of both producing and consuming messages.
+
+
+#### KafkaClientHandle
+
+Demonstrates how to wrap the Confluent.Kafka producer in a class that can be registered as
+a singleton service, including how to bind client configuration from an injected IConfiguration
+instance to a Confluent.Kafka.ProducerConfig object.
+
+Important Note: You should not create new client instances frequently (for example in the
+body of a request handler) - doing this is enormously expensive. Instead, you should create a
+single client instance that lives for the lifetime of the application as per this example.
+The .NET Kafka client is completely thread safe and a single instance is capable of serving
+the requirements tens or hundreds of thousands of concurrent web requests.
+
+
+#### KafkaDependentProducer
+
+Demonstrates how to produce messages with different types using a single producer instance.
+This is more efficient than creating more than one producer instance.
+
+
+#### HomeController
+
+Demonstrates how utilize a previously registered KafkaDependentProducer singleton service in
+a controller.
+
+
+#### RequestTimerMiddleware
+
+Demonstrates how utilize a KafkaDependentProducer service in a middleware component that
+measures how long a web request takes to handle, and logs the information to Kafka.
+
+
+#### RequestTimeConsumer
+
+Demonstrates how to run a Confluent.Kafka consumer as an IHostedService.
+