Merge pull request #209 from alfa-laboratory/feature/kafka

Feature/kafka
alfa-laboratory · Dec 13, 2021 · 5cb899e · 5cb899e
2 parents ffc36d0 + 861fae5
commit 5cb899e
Show file tree

Hide file tree

Showing 3 changed files with 45 additions and 4 deletions.
diff --git a/examples/Molder.Kafka.Example/appsettings.json b/examples/Molder.Kafka.Example/appsettings.json
@@ -8,7 +8,10 @@
           "AutoOffsetReset": "Earliest",
           "AutoCommitIntervalMs": 5000,
           "SessionTimeoutMs": 6000,
-          "EnableAutoCommit": true
+          "EnableAutoCommit": true,
+          "SaslUsername": "username",
+          "SaslPassword": "password",
+          "SaslMechanism": "OAuthBearer"
         },
         "Topic": "test-topic",
         "Name": "test"

diff --git a/src/Molder.Kafka/Molder.Kafka.csproj b/src/Molder.Kafka/Molder.Kafka.csproj
@@ -11,7 +11,7 @@
         <IsPackable>true</IsPackable>
         <Description>Library for work with Kafka</Description>
         <LangVersion>9</LangVersion>
-        <PackageVersion>1.0.0</PackageVersion>
+        <PackageVersion>1.0.1</PackageVersion>
         <AssemblyVersion>$(PackageVersion)</AssemblyVersion>
         <Nullable>disable</Nullable>
     </PropertyGroup>

diff --git a/src/Molder.Kafka/Steps/KafkaSteps.cs b/src/Molder.Kafka/Steps/KafkaSteps.cs
@@ -1,7 +1,10 @@
-using FluentAssertions;
+using Confluent.Kafka;
+using FluentAssertions;
 using Molder.Controllers;
+using Molder.Extensions;
 using Molder.Kafka.Models;
 using TechTalk.SpecFlow;
+using TechTalk.SpecFlow.Assist;
 
 namespace Molder.Kafka.Steps
 {
@@ -16,7 +19,42 @@ public KafkaSteps(VariableController variableController)
         {
             this.variableController = variableController;
         }
-
+
+        #region Transformations
+        /// <summary>
+        /// Трансформация параметров подключения к Kafka.
+        /// </summary>
+        /// <param name="table">Параметры подключения.</param>
+        /// <returns>Параметры подключения к Kafka.</returns>
+        [StepArgumentTransformation]
+        public ConsumerConfig GetKafkaParametersFromTable(Table table)
+        {
+            return table.ReplaceWith(variableController).CreateInstance<ConsumerConfig>();
+        }
+
+        #endregion
+        #region Connections
+        /// <summary>
+        /// Подключение к Kafka.
+        /// </summary>
+        /// <param name="name">Название подключения.</param>
+        /// <param name="topic">Название топика.</param>
+        /// <param name="consumerConfig">Параметры подключения.</param>
+        [Given(@"я создаю параметры для подключения к kafka c именем ""(.+)"" и топиком ""(.+)"":")]
+        public void ConnectToKafka_Kafka(string name, string topic, ConsumerConfig consumerConfig)
+        {
+            KafkaSettings.Settings.Should().NotContainKey(name,
+                $"параметров подлючения к kafka с именем \"{name}\" не существует");
+
+            KafkaSettings.Settings.Add(name, new Settings()
+            {
+                Config = consumerConfig,
+                Name = name,
+                Topic = topic
+            });
+        }
+        #endregion
+
         [StepDefinition(@"запустить обработчик очереди kafka c именем \""(.+)\""")]
         public void Run(string name)
         {