v1.0.0

Author: lefebsy

.dockerignore

@@ -0,0 +1,6 @@
+.vscode
+.git
+.cache
+**/obj/
+**/bin/
+appsettings.Development.json

.gitignore

@@ -0,0 +1,8 @@
+.vscode
+.vscode/settings.json
+obj
+bin
+out
+**/obj/
+**/bin/
+*.bak

BasicAuth/BasicAuthAttribute.cs

@@ -0,0 +1,14 @@
+using System;
+using Microsoft.AspNetCore.Mvc;
+
+namespace Json2Kafka.BasicAuth
+{
+    [AttributeUsage(AttributeTargets.Class | AttributeTargets.Method)]
+    public class BasicAuthAttribute : TypeFilterAttribute
+    {
+        public BasicAuthAttribute(string realm = @"My Realm") : base(typeof(BasicAuthFilter))
+        {
+            Arguments = new object[] { realm };
+        }
+    }
+}

BasicAuth/BasicAuthFilter.cs

@@ -0,0 +1,67 @@
+using System;
+using System.Net;
+using System.Net.Http.Headers;
+using System.Text;
+using Json2Kafka.Services;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.AspNetCore.Mvc.Filters;
+using Microsoft.Extensions.DependencyInjection;
+
+namespace Json2Kafka.BasicAuth
+{
+    public class BasicAuthFilter : IAuthorizationFilter
+    {
+        private readonly string _realm;
+        public BasicAuthFilter(string realm)
+        {
+            _realm = realm;
+            if (string.IsNullOrWhiteSpace(_realm))
+            {
+                throw new ArgumentNullException(nameof(realm), @"Please provide a non-empty realm value.");
+            }
+        }
+        public void OnAuthorization(AuthorizationFilterContext context)
+        {
+            try
+            {
+                string authHeader = context.HttpContext.Request.Headers["Authorization"];
+                if (authHeader != null)
+                {
+                    var authHeaderValue = AuthenticationHeaderValue.Parse(authHeader);
+                    if (authHeaderValue.Scheme.Equals(AuthenticationSchemes.Basic.ToString(), StringComparison.OrdinalIgnoreCase))
+                    {
+                        var credentials = Encoding.UTF8
+                                            .GetString(Convert.FromBase64String(authHeaderValue.Parameter ?? string.Empty))
+                                            .Split(':', 2);
+                        if (credentials.Length == 2)
+                        {
+                            if (IsAuthorized(context, credentials[0], credentials[1]))
+                            {
+                                return;
+                            }
+                        }
+                    }
+                }
+
+                ReturnUnauthorizedResult(context);
+            }
+            catch (FormatException)
+            {
+                ReturnUnauthorizedResult(context);
+            }
+        }
+
+        public bool IsAuthorized(AuthorizationFilterContext context, string username, string password)
+        {
+            var userService = context.HttpContext.RequestServices.GetRequiredService<IUserService>();
+            return userService.IsValidUser(username, password);
+        }
+
+        private void ReturnUnauthorizedResult(AuthorizationFilterContext context)
+        {
+            // Return 401 and a basic authentication challenge (causes browser to show login dialog)
+            context.HttpContext.Response.Headers["WWW-Authenticate"] = $"Basic realm=\"{_realm}\"";
+            context.Result = new UnauthorizedResult();
+        }
+    }
+}

Controllers/MsgController.cs

@@ -0,0 +1,51 @@
+using System;
+using System.Threading.Tasks;
+using Microsoft.AspNetCore.Mvc;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+using Json2Kafka.Services;
+using Json2Kafka.BasicAuth;
+
+namespace Json2Kafka.Controllers
+{
+    // Ajoute une route vide pour afficher l'aide de la méthode GET par défaut en cas d'affichage de la racine sur un navigateur
+    [Route("")]
+    [Route("api/[controller]")]
+    [ApiController]
+    public class MsgController : ControllerBase
+    {
+
+        private ProducerService _Producer;
+
+        // Constructeur du controller de l'API, instanciation du producer Kafka
+        public MsgController(IConfiguration Configuration) {
+            _Producer = ProducerService.GetInstance(Configuration);
+        }
+
+        // GET: / or /api/[controller] 
+        // Ne fait rien à part afficher une info sur la méthode POST à utiliser
+        [HttpGet]
+        public ActionResult<Object> GetMessage() => Content("Please use POST method on /api/msg - UTF8 JSON object mandatory");
+
+
+
+        // POST: /api/[controller]
+        // Méthode asynchrone
+        // le message vient du body de la request POST http.
+        // le message doit être du JSON UTF8 : {"timestamp": 123,"info": "myData","myKey":"myValue"}
+        // si message est différent d'un json la méthode retourne une erreur http "415 Unsupported Media Type"
+        // si la requête est incorrecte la méthode retourne "400 Bad Request"
+        // si BasicAuth est activé dans la configuration et que le login ou mot de passe sont incorects la méthode retourne une erreur "401 Unauthorized"
+        // si la requête est correcte la réponse de kafka est envoyée
+        // si kafka est injoignable le producer stock et attend le retour de Kafka (time out kafka par défaut 5 minutes / 300000ms)
+        [HttpPost]
+        [Route("api/[controller]")] // oblige à utiliser cette route : /api/msg
+        [HttpGet("basic")]
+        [BasicAuth("thisServiceNotKafka")] 
+        public async Task<ActionResult<Object>> PostMessage(Object message)
+        {
+            var deliveryResult = await _Producer.ProduceAwait(message); // send the message to kafka
+            return Created("Kafka", deliveryResult); // return http 201 created to the requester with a DeliveryResult or ProduceException in case of problem
+        }
+    }
+}

Dockerfile

@@ -0,0 +1,24 @@
+# https://hub.docker.com/_/microsoft-dotnet-core
+FROM mcr.microsoft.com/dotnet/core/sdk:3.1-alpine AS build
+WORKDIR /source
+
+# copy csproj and restore as distinct layers
+COPY *.csproj .
+RUN dotnet restore -r linux-musl-x64
+
+# copy everything else and build app
+COPY . .
+RUN dotnet publish -c release -o /app -r linux-musl-x64 --self-contained true /p:PublishTrimmed=true /p:PublishReadyToRun=true /p:PublishReadyToRunShowWarnings=true
+#RUN dotnet publish -c release -o /app -r linux-musl-x64 --self-contained false
+
+# final stage/image
+FROM mcr.microsoft.com/dotnet/core/runtime-deps:3.1-alpine
+#FROM mcr.microsoft.com/dotnet/core/aspnet:3.1-alpine
+LABEL author="Lefebsy" \
+      info="DotnetCore3.1 - Simple REST Json webservice feeding a kafka producer"
+WORKDIR /app
+COPY --from=build /app ./
+
+EXPOSE 80/tcp
+#name of the csproj
+ENTRYPOINT ["./Json2Kafka"] 

Json2Kafka.csproj

@@ -0,0 +1,11 @@
+<Project Sdk="Microsoft.NET.Sdk.Web">
+
+  <PropertyGroup>
+    <TargetFramework>netcoreapp3.1</TargetFramework>
+  </PropertyGroup>
+
+  <ItemGroup>
+    <PackageReference Include="Confluent.Kafka" Version="1.4.2" />
+  </ItemGroup>
+
+</Project>

Program.cs

@@ -0,0 +1,25 @@
+using Microsoft.AspNetCore.Hosting;
+using Microsoft.Extensions.Hosting;
+
+namespace Json2Kafka
+{
+    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>();
+                });
+
+    }
+
+
+}

Properties/launchSettings.json

@@ -0,0 +1,30 @@
+{
+  "$schema": "http://json.schemastore.org/launchsettings.json",
+  "iisSettings": {
+    "windowsAuthentication": false,
+    "anonymousAuthentication": true,
+    "iisExpress": {
+      "applicationUrl": "http://localhost:55302",
+      "sslPort": 44304
+    }
+  },
+  "profiles": {
+    "IIS Express": {
+      "commandName": "IISExpress",
+      "launchBrowser": true,
+      "launchUrl": "message",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    },
+    "Json2Kafka": {
+      "commandName": "Project",
+      "launchBrowser": true,
+      "launchUrl": "message",
+      "applicationUrl": "https://localhost:5001;http://localhost:5000",
+      "environmentVariables": {
+        "ASPNETCORE_ENVIRONMENT": "Development"
+      }
+    }
+  }
+}

Services/ProducerService.cs

@@ -0,0 +1,97 @@
+using System;
+using System.Threading.Tasks;
+using Confluent.Kafka;
+using Microsoft.Extensions.Configuration;
+using Microsoft.Extensions.Logging;
+
+namespace Json2Kafka.Services
+{
+    class ProducerService
+    {
+        private static IProducer<Null, string> _Producer; // singleton du producer
+        private static Boolean _SetupDone = false; //pour ne pas rejouer x fois la config
+        private static ProducerConfig _ProducerConfig; // configuration du producer (depuis les variables d'environnement)
+        private static string _Topic; // (depuis les variables d'environnement)
+        private static ILogger _logger;
+        
+        
+        
+        // Constructeur du singleton et de la gestion de l'obtention de son instance (paramètres interdits dans le constructeur, donc Setup depuis une méthode séparée)
+        private static readonly ProducerService _ProducerServiceInstance = new ProducerService();
+        public static ProducerService GetInstance(IConfiguration Configuration) {
+            if (!_SetupDone)
+            {
+                // création de la configuration Kafka à partir des infos récupèrées
+                // dans appsettings.json qui peuvent être surchargées par
+                // les variables d'environements pour tourner en container
+
+                // configs obligatoires
+                _Topic = Configuration["Topic"];            
+                _ProducerConfig = new ProducerConfig {
+                    ClientId = Configuration["ClientId"],
+                    CompressionType = CompressionType.Snappy, //Compression active par défaut, faible charge CPU et 0.5x de traffic et empreinte stockage sur kafka
+                    BootstrapServers = Configuration["BootstrapServers"],
+                    EnableSslCertificateVerification = bool.Parse( Configuration["EnableSslCertificateVerification"] ),
+                    EnableIdempotence = bool.Parse( Configuration["EnableIdempotence"] )
+                };
+                // configs optionnelles
+                if ("" != Configuration["SaslPassword"]) _ProducerConfig.SaslPassword = Configuration["SaslPassword"];
+                if ("" != Configuration["SaslUsername"]) _ProducerConfig.SaslUsername = Configuration["SaslUsername"];
+                if ("" != Configuration["SslCaLocation"]) _ProducerConfig.SslCaLocation = Configuration["SslCaLocation"];
+                if ("" != Configuration["SaslMechanism"]) _ProducerConfig.SaslMechanism = (SaslMechanism)int.Parse(Configuration["SaslMechanism"]);
+                if ("" != Configuration["SecurityProtocol"]) _ProducerConfig.SecurityProtocol = (SecurityProtocol)int.Parse(Configuration["SecurityProtocol"]);
+
+                _Producer = new ProducerBuilder<Null, string>(_ProducerConfig).Build();
+                _logger = LoggerFactory.Create(builder => builder.AddConsole().AddConfiguration(Configuration.GetSection("Logging"))).CreateLogger("Json2Kafka.services.ProducerService");
+                _SetupDone = true;
+            }
+
+            return _ProducerServiceInstance;
+        }
+
+        // méthode principale pour écrire dans Kafka
+        public async Task<object> ProduceAwait(object msg)
+        {
+            try
+            {
+                _logger.LogInformation($"Sending message to producer [{msg}]");
+                return await _Producer.ProduceAsync(_Topic, new Message<Null, string> { Value=msg.ToString() });
+            }
+            catch (ProduceException<Null, string> e)
+            {
+                //Console.WriteLine($"Delivery failed: {e.Error.Reason}");
+                _logger.LogError($"Delivery failed: {e.Error.Reason}");
+                return e;
+            }
+        } 
+        // méthode en cours de test
+        private void Produce(object msg)
+        {
+            Action<DeliveryReport<Null, string>> handler = r => 
+                Console.WriteLine(!r.Error.IsError
+                    ? $"Delivered message to {r.TopicPartitionOffset}"
+                    : $"Delivery Error: {r.Error.Reason}");
+
+            _Producer.Produce(_Topic, new Message<Null, string> { Value = msg.ToString() }, handler);
+            
+            // wait for up to 10 seconds for any inflight messages to be delivered.
+            _Producer.Flush(TimeSpan.FromSeconds(10));
+            
+        }
+        // méthode en cours de test
+        private async Task ProduceAsync(object msg)
+        {
+            try
+            {
+                var dr = await _Producer.ProduceAsync(_Topic, new Message<Null, string> { Value=msg.ToString() });
+                Console.WriteLine($"Delivered '{dr.Value}' to '{dr.TopicPartitionOffset}'");
+            }
+            catch (ProduceException<Null, string> e)
+            {
+                Console.WriteLine($"Delivery failed: {e.Error.Reason}");
+            }            
+        }
+
+    }
+
+}