YAML

pom.xml

@@ -109,6 +109,7 @@ SOFTWARE.
     <maven-gpg-plugin.version>3.1.0</maven-gpg-plugin.version>
     <mockito-core.version>5.4.0</mockito-core.version>
     <mockito-junit-jupiter.version>5.4.0</mockito-junit-jupiter.version>
+    <snakeyaml.version>2.0</snakeyaml.version>
   </properties>
   <dependencies>
     <dependency>
@@ -140,6 +141,11 @@ SOFTWARE.
       <artifactId>xembly</artifactId>
       <version>${xembly.version}</version>
     </dependency>
+    <dependency>
+      <groupId>org.yaml</groupId>
+      <artifactId>snakeyaml</artifactId>
+      <version>${snakeyaml.version}</version>
+    </dependency>
     <dependency>
       <groupId>com.jcabi</groupId>
       <artifactId>jcabi-immutable</artifactId>

src/main/java/io/github/eocqrs/kafka/xml/XmlMapParams.java

@@ -28,6 +28,7 @@
 import org.cactoos.Scalar;
 import org.cactoos.io.ResourceOf;
 
+import java.util.Collections;
 import java.util.Locale;
 import java.util.Map;
 import java.util.regex.Pattern;
@@ -72,7 +73,7 @@ protected XmlMapParams(final XML config, final KfCustomer cust) {
    * A ctor that takes an Input and converts it to XMLDocument.
    *
    * @param resource Resource with settings.
-   * @param cust  Customer type.
+   * @param cust     Customer type.
    * @throws Exception When something went wrong.
    */
   protected XmlMapParams(final Input resource, final KfCustomer cust)
@@ -94,25 +95,27 @@ protected XmlMapParams(final String name, final KfCustomer cust)
 
   @Override
   public final Map<String, Object> value() throws Exception {
-    return new XMLDocument(this.configuration.toString())
-      .nodes("//%s/*".formatted(this.customer))
-      .stream()
-      .map(Object::toString)
-      .map(XMLDocument::new)
-      .map(xml -> xml.nodes("//*").get(0).node().getNodeName())
-      .collect(
-        Collectors.toMap(
-          name ->
-            XmlMapParams.CAPITALS
-              .matcher(name)
-              .replaceAll(".$1")
-              .toLowerCase(Locale.ROOT),
-          name ->
-            new TextXpath(
-              this.configuration,
-              "//%s".formatted(name)
-            ).toString()
+    return Collections.unmodifiableMap(
+      new XMLDocument(this.configuration.toString())
+        .nodes("//%s/*".formatted(this.customer))
+        .stream()
+        .map(Object::toString)
+        .map(XMLDocument::new)
+        .map(xml -> xml.nodes("//*").get(0).node().getNodeName())
+        .collect(
+          Collectors.toMap(
+            name ->
+              XmlMapParams.CAPITALS
+                .matcher(name)
+                .replaceAll(".$1")
+                .toLowerCase(Locale.ROOT),
+            name ->
+              new TextXpath(
+                this.configuration,
+                "//%s".formatted(name)
+              ).toString()
+          )
         )
-      );
+    );
   }
 }

src/main/java/io/github/eocqrs/kafka/yaml/KfYamlConsumerSettings.java

@@ -0,0 +1,48 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.github.eocqrs.kafka.yaml;
+
+import io.github.eocqrs.kafka.ConsumerSettings;
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.consumer.KafkaConsumer;
+
+/**
+ * Yaml configuration for consumer.
+ *
+ * @param <K> The key type.
+ * @param <X> The value type.
+ */
+@RequiredArgsConstructor
+public final class KfYamlConsumerSettings<K, X>
+  implements ConsumerSettings<K, X> {
+
+  /**
+   * YAML params.
+   */
+  private final YamlMapParams<K, X> params;
+
+  @Override
+  public KafkaConsumer<K, X> consumer() {
+    return new KafkaConsumer<>(this.params.value());
+  }
+}

src/main/java/io/github/eocqrs/kafka/yaml/KfYamlProducerSettings.java

@@ -0,0 +1,48 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.github.eocqrs.kafka.yaml;
+
+import io.github.eocqrs.kafka.ProducerSettings;
+import lombok.RequiredArgsConstructor;
+import org.apache.kafka.clients.producer.KafkaProducer;
+
+/**
+ * Yaml configuration for producer.
+ *
+ * @param <K> The key type.
+ * @param <X> The value type.
+ */
+@RequiredArgsConstructor
+public final class KfYamlProducerSettings<K, X>
+  implements ProducerSettings<K, X> {
+
+  /**
+   * YAML params.
+   */
+  private final YamlMapParams<K, X> params;
+
+  @Override
+  public KafkaProducer<K, X> producer() {
+    return new KafkaProducer<>(this.params.value());
+  }
+}

src/main/java/io/github/eocqrs/kafka/yaml/YamlMapParams.java

@@ -0,0 +1,92 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.github.eocqrs.kafka.yaml;
+
+import lombok.RequiredArgsConstructor;
+import org.cactoos.Input;
+import org.cactoos.Scalar;
+import org.cactoos.io.ResourceOf;
+import org.yaml.snakeyaml.Yaml;
+
+import java.io.InputStream;
+import java.util.Collections;
+import java.util.HashMap;
+import java.util.Map;
+
+/**
+ * This class converts a YAML source from
+ * kebab case keys into kafka specific keys Map.
+ *
+ * @param <K> The key type.
+ * @param <X> The value type.
+ */
+@RequiredArgsConstructor
+public class YamlMapParams<K, X> implements Scalar<Map<String, Object>> {
+
+  /**
+   * The config value.
+   */
+  private final Map<String, Object> value;
+
+  /**
+   * Filename ctor.
+   *
+   * @param name Config name
+   * @throws Exception When something went wrong
+   */
+  public YamlMapParams(final String name) throws Exception {
+    this(new ResourceOf(name));
+  }
+
+  /**
+   * Input ctor.
+   *
+   * @param input Input source
+   * @throws Exception When something went wrong
+   */
+  public YamlMapParams(final Input input) throws Exception {
+    this(input.stream());
+  }
+
+  /**
+   * Primary ctor.
+   *
+   * @param stream Input source
+   */
+  public YamlMapParams(final InputStream stream) {
+    this.value = new Yaml().load(stream);
+  }
+
+  @Override
+  public final Map<String, Object> value() {
+    final Map<String, Object> accum = new HashMap<>(0);
+    this.value
+      .forEach(
+        (key, val) -> accum.put(
+          key.replace('-', '.'),
+          val
+        )
+      );
+    return Collections.unmodifiableMap(accum);
+  }
+}

src/main/java/io/github/eocqrs/kafka/yaml/package-info.java

@@ -0,0 +1,26 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+/**
+ * YAML.
+ */
+package io.github.eocqrs.kafka.yaml;

src/test/java/io/github/eocqrs/kafka/yaml/KfYamlConsumerSettingsTest.java

@@ -0,0 +1,46 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.github.eocqrs.kafka.yaml;
+
+import io.github.eocqrs.kafka.consumer.KfConsumer;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Test case for {@link KfYamlProducerSettings}.
+ */
+class KfYamlConsumerSettingsTest {
+
+  @Test
+  void createsConsumerFromYamlConfiguration() {
+    assertDoesNotThrow(
+      () ->
+        new KfConsumer<>(
+          new KfYamlConsumerSettings<>(
+            new YamlMapParams<String, String>("consumer.yaml")
+          )
+        )
+    );
+  }
+}

src/test/java/io/github/eocqrs/kafka/yaml/KfYamlProducerSettingsTest.java

@@ -0,0 +1,47 @@
+/*
+ *  Copyright (c) 2022 Aliaksei Bialiauski, EO-CQRS
+ *
+ * Permission is hereby granted, free of charge, to any person obtaining a copy
+ * of this software and associated documentation files (the "Software"), to deal
+ * in the Software without restriction, including without limitation the rights
+ * to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
+ * copies of the Software, and to permit persons to whom the Software is
+ * furnished to do so, subject to the following conditions:
+ *
+ * The above copyright notice and this permission notice shall be included in all
+ * copies or substantial portions of the Software.
+ *
+ * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
+ * IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
+ * FITNESS FOR A PARTICULAR PURPOSE AND NON-INFRINGEMENT. IN NO EVENT SHALL THE
+ * AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
+ * LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
+ * OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN THE
+ * SOFTWARE.
+ */
+
+package io.github.eocqrs.kafka.yaml;
+
+import io.github.eocqrs.kafka.producer.KfProducer;
+import org.junit.jupiter.api.Test;
+
+import static org.junit.jupiter.api.Assertions.*;
+
+/**
+ * Test case for {@link KfYamlConsumerSettings}.
+ */
+class KfYamlProducerSettingsTest {
+
+  @Test
+  void createsProducerFromYaml() {
+    assertDoesNotThrow(
+      () -> new KfProducer<>(
+        new KfYamlProducerSettings<>(
+          new YamlMapParams<String, String>(
+            "producer.yaml"
+          )
+        )
+      )
+    );
+  }
+}