230807

.env

@@ -48,8 +48,8 @@ RABBITMQ_DEFAULT_PASS='admin123@'
 
 # Kafka
 #
-KAFKA_CLIENT_USERS='kafka'
-KAFKA_CLIENT_PASSWORDS='admin123@'
+KAFKA_DEFAULT_USERS='kafka'
+KAFKA_DEFAULT_PASSWORDS='admin123@'
 
 # Redis
 #

docker-compose.yml

@@ -33,8 +33,8 @@ services:
       MONITORING_INTERNAL_PASSWORD: ${MONITORING_INTERNAL_PASSWORD:-}
       RABBITMQ_DEFAULT_USER: ${RABBITMQ_DEFAULT_USER:-}
       RABBITMQ_DEFAULT_PASS: ${RABBITMQ_DEFAULT_PASS:-}
-      KAFKA_CLIENT_USERS: ${KAFKA_CLIENT_USERS:-}
-      KAFKA_CLIENT_PASSWORDS: ${KAFKA_CLIENT_PASSWORDS:-}
+      KAFKA_DEFAULT_USERS: ${KAFKA_DEFAULT_USERS:-}
+      KAFKA_DEFAULT_PASSWORDS: ${KAFKA_DEFAULT_PASSWORDS:-}
     networks:
       - yan
     depends_on:
@@ -127,7 +127,11 @@ services:
     ports:
       - 2181:2181
     environment:
-      - ALLOW_ANONYMOUS_LOGIN=yes
+      ZOO_ENABLE_AUTH: "yes"
+      ZOO_SERVER_USERS: ${KAFKA_DEFAULT_USERS:-}
+      ZOO_SERVER_PASSWORDS: ${KAFKA_DEFAULT_PASSWORDS:-}
+      ZOO_CLIENT_USER: ${KAFKA_DEFAULT_USERS:-}
+      ZOO_CLIENT_PASSWORD: ${KAFKA_DEFAULT_PASSWORDS:-}
     networks:
       - yan
     restart: unless-stopped
@@ -140,16 +144,23 @@ services:
       - 9093:9093
     environment:
       BITNAMI_DEBUG: "yes"
-      KAFKA_ENABLE_KRAFT: "no"
+      ALLOW_PLAINTEXT_LISTENER: "no"
       KAFKA_BROKER_ID: 1
+      KAFKA_ENABLE_KRAFT: "no"
       KAFKA_CFG_ZOOKEEPER_CONNECT: zookeeper:2181
-      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
-      KAFKA_CFG_LISTENERS: CLIENT://:9093,EXTERNAL://:9092
-      KAFKA_CFG_ADVERTISED_LISTENERS: CLIENT://kafka:9093,EXTERNAL://localhost:9092
+      KAFKA_CFG_ADVERTISED_LISTENERS: INTERNAL://kafka:9092,CLIENT://localhost:9093
+      KAFKA_CFG_LISTENERS: INTERNAL://:9092,CLIENT://:9093
+      KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP: INTERNAL:SASL_PLAINTEXT,CLIENT:SASL_PLAINTEXT
+      KAFKA_CFG_SASL_MECHANISM_INTER_BROKER_PROTOCOL: PLAIN
       KAFKA_CFG_INTER_BROKER_LISTENER_NAME: CLIENT
-      ALLOW_PLAINTEXT_LISTENER: "yes"
-      KAFKA_CLIENT_USERS: ${KAFKA_CLIENT_USERS:-}
-      KAFKA_CLIENT_PASSWORDS: ${KAFKA_CLIENT_PASSWORDS:-}
+      KAFKA_INTER_BROKER_LISTENER_NAME: INTERNAL
+      KAFKA_INTER_BROKER_USER: ${KAFKA_DEFAULT_USERS:-}
+      KAFKA_INTER_BROKER_PASSWORD: ${KAFKA_DEFAULT_PASSWORDS:-}
+      KAFKA_ZOOKEEPER_PROTOCOL: SASL_PLAINTEXT
+      KAFKA_ZOOKEEPER_USER: ${KAFKA_DEFAULT_USERS:-}
+      KAFKA_ZOOKEEPER_PASSWORD: ${KAFKA_DEFAULT_PASSWORDS:-}
+      KAFKA_CLIENT_USERS: ${KAFKA_DEFAULT_USERS:-}
+      KAFKA_CLIENT_PASSWORDS: ${KAFKA_DEFAULT_PASSWORDS:-}
     networks:
       - yan
     depends_on:
@@ -162,9 +173,14 @@ services:
     ports:
       - 8080:8080
     environment:
-      - KAFKA_CLUSTERS_0_NAME=local
-      - KAFKA_CLUSTERS_0_ZOOKEEPER=zookeeper:2181
-      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9093
+      KAFKA_CLUSTERS_0_NAME: local
+      KAFKA_CLUSTERS_0_ZOOKEEPER: zookeeper:2181
+      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:9092
+      KAFKA_CLUSTERS_0_PROPERTIES_SECURITY_PROTOCOL: SASL_PLAINTEXT
+      KAFKA_CLUSTERS_0_PROPERTIES_SASL_MECHANISM: PLAIN
+      KAFKA_CLUSTERS_0_PROPERTIES_CLIENT_DNS_LOOKUP: use_all_dns_ips
+      KAFKA_CLUSTERS_0_PROPERTIES_SASL_JAAS_CONFIG: "org.apache.kafka.common.security.plain.PlainLoginModule required username='${KAFKA_DEFAULT_USERS:-}' password='${KAFKA_DEFAULT_PASSWORDS:-}';"
+      KAFKA_CLUSTERS_0_DISABLELOGDIRSCOLLECTION: true
     networks:
       - yan
     depends_on:
@@ -179,7 +195,6 @@ networks:
 volumes:
   setup:
   elasticsearch:
-  es-ingest:
 
 # Run with
 #   docker-compose \

host/YANLib.HttpApi.Host/YANLib.HttpApi.Host.csproj

@@ -15,6 +15,10 @@
 	<ItemGroup>
 		<PackageReference Include="AspNetCore.HealthChecks.SqlServer" Version="6.0.2" />
 		<PackageReference Include="AspNetCore.HealthChecks.UI.Client" Version="6.0.5" />
+		<PackageReference Include="DotNetCore.CAP" Version="6.2.1" />
+		<PackageReference Include="DotNetCore.CAP.Dashboard" Version="6.2.1" />
+		<PackageReference Include="DotNetCore.CAP.Kafka" Version="6.2.1" />
+		<PackageReference Include="DotNetCore.CAP.MongoDB" Version="6.2.1" />
 		<PackageReference Include="Elastic.Apm.NetCoreAll" Version="1.22.0" />
 		<PackageReference Include="Serilog.AspNetCore" Version="6.1.0" />
 		<PackageReference Include="Serilog.Formatting.Compact" Version="1.1.0" />
@@ -27,7 +31,6 @@
 		<PackageReference Include="Volo.Abp.AspNetCore.Serilog" Version="6.0.3" />
 		<PackageReference Include="Volo.Abp.Caching.StackExchangeRedis" Version="6.0.3" />
 		<PackageReference Include="Volo.Abp.EntityFrameworkCore.SqlServer" Version="6.0.3" />
-		<PackageReference Include="Volo.Abp.EventBus.Kafka" Version="6.0.3" />
 		<PackageReference Include="Volo.Abp.EventBus.RabbitMQ" Version="6.0.3" />
 		<PackageReference Include="Volo.Abp.Http.Client" Version="6.0.3" />
 		<PackageReference Include="Volo.Abp.Swashbuckle" Version="6.0.3" />

host/YANLib.HttpApi.Host/YANLibHttpApiHostModule.cs

@@ -1,3 +1,4 @@
+using DotNetCore.CAP;
 using Elastic.Apm.DiagnosticSource;
 using Elastic.Apm.EntityFrameworkCore;
 using Elastic.Apm.NetCoreAll;
@@ -9,6 +10,7 @@
 using Microsoft.Extensions.DependencyInjection;
 using Microsoft.Extensions.Hosting;
 using Microsoft.OpenApi.Models;
+using MongoDB.Driver;
 using System;
 using System.Collections.Generic;
 using System.Linq;
@@ -43,7 +45,6 @@ namespace YANLib;
     typeof(AbpSwashbuckleModule),
     typeof(AbpEntityFrameworkCoreSqlServerModule),
     typeof(AbpCachingStackExchangeRedisModule),
-    //typeof(AbpEventBusKafkaModule),
     typeof(AbpEventBusRabbitMqModule),
     typeof(AbpHttpClientModule)
 )]
@@ -80,6 +81,40 @@ private static void ConfigureSwaggerServices(ServiceConfigurationContext context
             o.HideAbpEndpoints();
             o.EnableAnnotations();
         });
+
+        _ = context.Services.AddSingleton<IMongoClient>(new MongoClient(configuration["CAP:ConnectionString"]));
+
+        _ = context.Services.AddCap(c =>
+        {
+            _ = c.UseDashboard(o => o.PathMatch = "/cap");
+
+            _ = c.UseMongoDB(o =>
+            {
+                o.DatabaseName = configuration["CAP:DBName"];
+                o.DatabaseConnection = configuration["CAP:ConnectionString"];
+            });
+
+            _ = c.UseKafka(o =>
+            {
+                o.Servers = configuration["CAP:Kafka:Connections:Default:BootstrapServers"];
+
+                o.MainConfig.Add("security.protocol", "SASL_PLAINTEXT");
+                o.MainConfig.Add("sasl.mechanism", "PLAIN");
+
+                if (configuration["CAP:Kafka:Username"].IsNotWhiteSpaceAndNull())
+                {
+                    o.MainConfig.Add("sasl.username", configuration["CAP:Kafka:Username"]);
+                }
+
+                if (configuration["CAP:Kafka:Password"].IsNotWhiteSpaceAndNull())
+                {
+                    o.MainConfig.Add("sasl.password", configuration["CAP:Kafka:Password"]);
+                }
+            });
+
+            c.DefaultGroupName = configuration["Cap:DefaultGroupName"] ?? c.DefaultGroupName;
+            c.FailedRetryCount = configuration["Cap:FailedRetryCount"].ToInt(0);
+        });
     }
 
     private void ConfigureLocalization() => Configure<AbpLocalizationOptions>(o =>
@@ -140,7 +175,7 @@ public override void OnApplicationInitialization(ApplicationInitializationContex
         _ = app.UseCorrelationId();
         _ = app.UseStaticFiles();
         _ = app.UseRouting();
-        _ = app.UseCors();
+        //_ = app.UseCors();
         _ = app.UseAuthentication();
         _ = app.UseUnitOfWork();
         _ = app.UseAuthorization();
@@ -164,5 +199,6 @@ public override void OnApplicationInitialization(ApplicationInitializationContex
         });
 
         _ = app.UseConfiguredEndpoints();
+        _ = app.UseCapDashboard();
     }
 }

host/YANLib.HttpApi.Host/appsettings.Development.json

@@ -19,17 +19,18 @@
       "ExchangeName": "YanlibDev"
     }
   },
-  "Kafka": {
-    "Connections": {
-      "Default": {
-        "BootstrapServers": "192.168.1.7:9092"
+  "CAP": {
+    "ConnectionString": "mongodb://mongo:admin123%[email protected]:27017/?authSource=admin",
+    "DBName": "yanlib",
+    "DefaultGroupName": "yanlib.dev",
+    "Kafka": {
+      "Connections": {
+        "Default": {
+          "BootstrapServers": "127.0.0.1:9093"
+        }
       },
       "Username": "kafka",
       "Password": "admin123@"
-    },
-    "EventBus": {
-      "GroupId": "YanlibDev",
-      "TopicName": "yan.lib"
     }
   },
   "Elasticsearch": {

host/YANLib.HttpApi.Host/appsettings.Production.json

@@ -19,17 +19,18 @@
       "ExchangeName": "YanlibProd"
     }
   },
-  "Kafka": {
-    "Connections": {
-      "Default": {
-        "BootstrapServers": "localhost:9092"
+  "CAP": {
+    "ConnectionString": "mongodb://mongo:admin123%40@mongodb:27017/?authSource=admin",
+    "DBName": "yanlib",
+    "DefaultGroupName": "yanlib.prod",
+    "Kafka": {
+      "Connections": {
+        "Default": {
+          "BootstrapServers": "kafka:9093"
+        }
       },
       "Username": "kafka",
       "Password": "admin123@"
-    },
-    "EventBus": {
-      "GroupId": "YanlibProd",
-      "TopicName": "yan.lib"
     }
   },
   "Elasticsearch": {

src/YANLib.Application.Contracts/EsIndexs/DeveloperIndex.cs → src/YANLib.Application.Contracts/EsIndexes/DeveloperIndex.cs

@@ -3,7 +3,7 @@
 using System.Collections.Generic;
 using YANLib.Responses;
 
-namespace YANLib.EsIndexs;
+namespace YANLib.EsIndexes;
 
 public sealed class DeveloperIndex
 {

src/YANLib.Application.Contracts/EsServices/IDeveloperEsService.cs

@@ -1,6 +1,6 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
-using YANLib.EsIndexs;
+using YANLib.EsIndexes;
 
 namespace YANLib.EsServices;
 

src/YANLib.Application/EsServices/DeveloperEsService.cs

@@ -5,7 +5,7 @@
 using System.Collections.Generic;
 using System.Linq;
 using System.Threading.Tasks;
-using YANLib.EsIndexs;
+using YANLib.EsIndexes;
 using YANLib.Utilities;
 using static System.Threading.Tasks.Task;
 using static YANLib.YANLibConsts;

src/YANLib.Application/Handlers/AdjustCertificateHandler.cs

@@ -2,7 +2,7 @@
 using System.Collections.Generic;
 using System.Threading.Tasks;
 using Volo.Abp.EventBus.Distributed;
-using YANLib.Etos;
+using YANLib.RabbitMq.Etos;
 using YANLib.Requests;
 using YANLib.Services;
 

src/YANLib.Application/Mappers/CertificateMapper.cs

@@ -1,7 +1,8 @@
 using AutoMapper;
 using Volo.Abp.AutoMapper;
-using YANLib.Etos;
+using YANLib.Kafka.Etos;
 using YANLib.Models;
+using YANLib.RabbitMq.Etos;
 using YANLib.Requests;
 using YANLib.Responses;
 

src/YANLib.Application/Mappers/DeveloperMapper.cs

@@ -1,6 +1,6 @@
 using AutoMapper;
 using Volo.Abp.AutoMapper;
-using YANLib.EsIndexs;
+using YANLib.EsIndexes;
 using YANLib.Models;
 using YANLib.Requests;
 using YANLib.Responses;