Merge pull request #392 from eo-cqrs/fake-consumer

Fake consumer records

src/main/java/io/github/eocqrs/kafka/Consumer.java

@@ -68,11 +68,14 @@ public interface Consumer<K, X> extends Closeable {
    * @param topic   topic to poll
    * @param timeout max time to wait
    * @return Records.
+   * @throws Exception When something went wrong.
    */
-  ConsumerRecords<K, X> records(String topic, Duration timeout);
+  ConsumerRecords<K, X> records(String topic, Duration timeout)
+    throws Exception;
 
   /**
    * Unsubscribe.
+   *
    * @throws Exception When something went wrong.
    */
   void unsubscribe() throws Exception;

src/main/java/io/github/eocqrs/kafka/fake/DatasetDirs.java

@@ -74,6 +74,9 @@ public Directives value() throws Exception {
       .set(this.key)
       .up()
       .addIf("value")
-      .set(this.data.dataized().dataize());
+      .set(this.data.dataized().dataize())
+      .up()
+      .addIf("seen")
+      .set("false");
   }
 }

src/main/java/io/github/eocqrs/kafka/fake/FkConsumer.java

@@ -33,16 +33,17 @@
 import java.time.LocalDateTime;
 import java.util.Collection;
 import java.util.UUID;
+/*
+ * @todo #303:90m/DEV fake consumer is not support various data types
+ */
 
 /**
  * Fake Consumer.
  *
- * @param <K> The key
- * @param <X> The value
  * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
  * @since 0.0.0
  */
-public final class FkConsumer<K, X> implements Consumer<K, X> {
+public final class FkConsumer implements Consumer<Object, String> {
 
   /**
    * Consumer id.
@@ -100,13 +101,32 @@ public void subscribe(final ConsumerRebalanceListener listener,
   }
 
   /*
-   * @todo #54:60m/DEV Fake records is not implemented
+   * @todo #303:45m/DEV records timeout is not implemented
    */
   @Override
-  public ConsumerRecords<K, X> records(
+  public ConsumerRecords<Object, String> records(
     final String topic, final Duration timeout
-  ) {
-    throw new UnsupportedOperationException("#records()");
+  ) throws Exception {
+    this.broker.with(new SubscribeDirs(topic, this.id).value());
+    final ConsumerRecords<Object, String> records =
+      new FkRecords(
+        topic,
+        this.broker.data(
+          ("broker/topics/topic[name = '%s']/datasets"
+            + "/dataset[seen = 'false']/value/text()"
+          ).formatted(
+            topic
+          )
+        )
+      ).value();
+    records.forEach(rec -> {
+      try {
+        this.broker.with(new SeenDirs(topic, rec.value()).value());
+      } catch (final Exception ex) {
+        throw new IllegalStateException(ex);
+      }
+    });
+    return records;
   }
 
   @Override

src/main/java/io/github/eocqrs/kafka/fake/FkRecords.java

@@ -0,0 +1,98 @@
+/*
+ *  Copyright (c) 2023 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.fake;
+
+import org.apache.kafka.clients.consumer.ConsumerRecord;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
+import org.apache.kafka.common.TopicPartition;
+import org.cactoos.Scalar;
+import org.cactoos.list.ListOf;
+
+import java.util.Collection;
+import java.util.HashMap;
+import java.util.List;
+import java.util.Map;
+
+/**
+ * Fake Records.
+ *
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.3.5
+ */
+public final class FkRecords implements
+  Scalar<ConsumerRecords<Object, String>> {
+
+  /*
+   * @todo #303:45m/DEV multi partitioning is not supported
+   */
+  /**
+   * Default Partition.
+   */
+  private static final int DEFAULT_PARTITION = 0;
+  /**
+   * Zero Offset.
+   */
+  private static final long ZERO_OFFSET = 0L;
+  /**
+   * Records related to a topic.
+   */
+  private final String topic;
+  /**
+   * Dataset to transform.
+   */
+  private final Collection<String> datasets;
+
+  /**
+   * Ctor.
+   *
+   * @param tpc  Topic
+   * @param dtst Dataset to transform
+   */
+  public FkRecords(
+    final String tpc,
+    final Collection<String> dtst
+  ) {
+    this.topic = tpc;
+    this.datasets = dtst;
+  }
+
+  @Override
+  public ConsumerRecords<Object, String> value() throws Exception {
+    final Map<TopicPartition, List<ConsumerRecord<Object, String>>> part
+      = new HashMap<>(0);
+    final List<ConsumerRecord<Object, String>> recs = new ListOf<>();
+    this.datasets.forEach(
+      dataset -> recs.add(
+        new ConsumerRecord<>(
+          this.topic,
+          DEFAULT_PARTITION,
+          ZERO_OFFSET,
+          null,
+          dataset
+        )
+      )
+    );
+    part.put(new TopicPartition(this.topic, DEFAULT_PARTITION), recs);
+    return new ConsumerRecords<>(part);
+  }
+}

src/main/java/io/github/eocqrs/kafka/fake/SeenDirs.java

@@ -0,0 +1,69 @@
+/*
+ *  Copyright (c) 2023 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.fake;
+
+import org.cactoos.Scalar;
+import org.xembly.Directives;
+
+/**
+ * Seen Directives.
+ *
+ * @author Aliaksei Bialiauski (abialiauski.dev@gmail.com)
+ * @since 0.3.5
+ */
+public final class SeenDirs implements Scalar<Directives> {
+
+  /**
+   * Topic.
+   */
+  private final String topic;
+  /**
+   * Fetched value.
+   */
+  private final String fetched;
+
+  /**
+   * Ctor.
+   *
+   * @param tpc Topic
+   * @param ftcd Fetched value
+   */
+  public SeenDirs(final String tpc, final String ftcd) {
+    this.topic = tpc;
+    this.fetched = ftcd;
+  }
+
+  @Override
+  public Directives value() throws Exception {
+    return new Directives()
+      .xpath(
+        ("broker/topics/topic[name = '%s']/datasets/dataset[value = '%s'"
+          + " and seen = 'false']/seen/text()")
+          .formatted(
+            this.topic,
+            this.fetched
+          )
+      )
+      .set("true");
+  }
+}

src/test/java/io/github/eocqrs/kafka/fake/DatasetDirsTest.java

@@ -17,7 +17,8 @@ final class DatasetDirsTest {
   void dirsInRightFormat() throws Exception {
     final String directives = "XPATH \"broker/topics/topic[name = '']\";ADDIF "
       + "\"datasets\";ADD \"dataset\";ADD \"partition\";\n"
-      + "4:SET \"0\";UP;ADDIF \"key\";SET \"test\";UP;ADDIF \"value\";SET \"\";";
+      + "4:SET \"0\";UP;ADDIF \"key\";SET \"test\";UP;ADDIF \"value\";SET \"\";"
+      + "UP;ADDIF \"seen\";SET \"false\";";
     MatcherAssert.assertThat(
       "Directives in the right format",
       new DatasetDirs<>(