#303

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/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

@@ -42,11 +42,32 @@
 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

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

@@ -33,9 +33,21 @@
  */
 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;

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

@@ -26,9 +26,12 @@
 
 import com.jcabi.log.Logger;
 import io.github.eocqrs.kafka.Consumer;
+import io.github.eocqrs.kafka.Producer;
+import io.github.eocqrs.kafka.data.KfData;
 import io.github.eocqrs.xfake.InFile;
 import io.github.eocqrs.xfake.Synchronized;
 import org.apache.kafka.clients.consumer.ConsumerRebalanceListener;
+import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.common.TopicPartition;
 import org.cactoos.list.ListOf;
 import org.hamcrest.MatcherAssert;
@@ -43,12 +46,10 @@
 import java.io.IOException;
 import java.time.Duration;
 import java.util.Collection;
-import java.util.Map;
+import java.util.List;
 import java.util.UUID;
 import java.util.logging.Level;
 
-import static org.junit.jupiter.api.Assertions.assertThrows;
-
 /**
  * Test case for {@link FkConsumer}.
  *
@@ -76,8 +77,8 @@ void setUp() throws Exception {
 
   @Test
   void createsWithMockBroker(@Mock final FkBroker mock) throws IOException {
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(UUID.randomUUID(), mock);
+    final Consumer<Object, String> consumer =
+      new FkConsumer(UUID.randomUUID(), mock);
     MatcherAssert.assertThat(
       "Fake consumer creates with mock broker",
       consumer,
@@ -88,8 +89,8 @@ void createsWithMockBroker(@Mock final FkBroker mock) throws IOException {
 
   @Test
   void creates() throws IOException {
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         UUID.randomUUID(),
         this.broker
       );
@@ -107,8 +108,8 @@ void subscribesToTopics() throws Exception {
     final UUID uuid = UUID.fromString("4b9d33f3-662f-41cf-a500-6169779e802a");
     final FkBroker with = this.broker
       .with(new TopicDirs(topic).value());
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         uuid,
         with
       );
@@ -143,8 +144,8 @@ void subscribesWithRebalanceListener() throws Exception {
     final UUID uuid = UUID.fromString("e343a512-9d02-40e8-92b6-1538014d3975");
     final FkBroker with = this.broker
       .with(new TopicDirs(topic).value());
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         uuid,
         with
       );
@@ -199,8 +200,8 @@ void subscribesWithVarArgs() throws Exception {
     final String topic = "varargs.test";
     final FkBroker with = this.broker
       .with(new TopicDirs(topic).value());
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         UUID.randomUUID(),
         with
       );
@@ -209,22 +210,23 @@ void subscribesWithVarArgs() throws Exception {
   }
 
   @Test
-  void throwsIfNull() {
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+  void throwsIfNull() throws IOException {
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         UUID.randomUUID(),
         this.broker
       );
     Assertions.assertThrows(
       IllegalStateException.class,
       () -> consumer.subscribe((String) null)
     );
+    consumer.close();
   }
 
   @Test
   void throwsIfNullWithListener() {
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         UUID.randomUUID(),
         this.broker
       );
@@ -246,8 +248,8 @@ public void onPartitionsAssigned(final Collection<TopicPartition> collection) {
   void unsubscribes() throws Exception {
     final String topic = "unsubscribe.test";
     final UUID uuid = UUID.randomUUID();
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
         uuid,
         this.broker
       );
@@ -283,13 +285,13 @@ void unsubscribesWithSecondConsumerExisting() throws Exception {
     final String topic = "unsubscribes.with.second.consumer.existing";
     final UUID firstID = UUID.fromString("f3000fb7-b9fb-42d0-8210-f09a58c44a1f");
     final UUID secondID = UUID.fromString("69a4cd5a-afdb-456c-9ade-658569f52d7b");
-    final Consumer<String, String> first =
-      new FkConsumer<>(
+    final Consumer<Object, String> first =
+      new FkConsumer(
         firstID,
         this.broker
       );
-    final Consumer<String, String> second =
-      new FkConsumer<>(
+    final Consumer<Object, String> second =
+      new FkConsumer(
         secondID,
         this.broker
       );
@@ -324,24 +326,58 @@ void unsubscribesWithSecondConsumerExisting() throws Exception {
   }
 
   @Test
-  void createsFakeConsumer() {
-    final FkConsumer<String, String> consumer =
-      new FkConsumer<>(UUID.randomUUID(), this.broker);
-    MatcherAssert.assertThat(consumer, Matchers.is(Matchers.notNullValue()));
-    assertThrows(
-      UnsupportedOperationException.class,
-      () -> consumer.records("123", Duration.ofMillis(100L))
+  void pollsRecordsWithoutException() throws Exception {
+    final Consumer<Object, String> consumer =
+      new FkConsumer(
+        UUID.randomUUID(),
+        this.broker
+      );
+    Assertions.assertDoesNotThrow(
+      () ->
+        consumer.records("test", Duration.ofSeconds(5L))
+    );
+    consumer.close();
+  }
+
+  @Test
+  void pollsRecords() throws Exception {
+    final String topic = "test";
+    final Consumer<Object, String> consumer =
+      new FkConsumer(UUID.randomUUID(),
+        this.broker
+          .with(new TopicDirs(topic).value())
+      );
+    final Producer<String, String> producer =
+      new FkProducer<>(UUID.randomUUID(), this.broker);
+    producer.send("test2", new KfData<>("test2", topic, 0));
+    producer.send("test.new", new KfData<>("test-data.new", topic, 0));
+    producer.send("test.new", new KfData<>("test-data.new", topic, 0));
+    final ConsumerRecords<Object, String> first =
+      consumer.records(topic, Duration.ofSeconds(1L));
+    final ConsumerRecords<Object, String> second =
+      consumer.records(topic, Duration.ofSeconds(1L));
+    final List<String> datasets = new ListOf<>();
+    first.forEach(rec -> datasets.add(rec.value()));
+    MatcherAssert.assertThat(
+      "First datasets in right format",
+      datasets,
+      Matchers.contains("test2", "test-data.new", "test-data.new")
     );
-    assertThrows(
-      UnsupportedOperationException.class,
-      () -> consumer.records("123", Duration.ofMillis(100L))
+    datasets.clear();
+    second.forEach(rec -> datasets.add(rec.value()));
+    MatcherAssert.assertThat(
+      "Second datasets are empty",
+      datasets.isEmpty(),
+      Matchers.equalTo(true)
     );
+    producer.close();
+    consumer.close();
   }
 
   @Test
   void closesWithoutException() {
-    final Consumer<String, String> consumer =
-      new FkConsumer<>(UUID.randomUUID(), this.broker);
+    final Consumer<Object, String> consumer =
+      new FkConsumer(UUID.randomUUID(), this.broker);
     Assertions.assertDoesNotThrow(consumer::close);
   }