From 60697e8f8e0e2dbd3d5745d55f0c037744081b28 Mon Sep 17 00:00:00 2001
From: Andy Coates <big-andy-coates@users.noreply.github.com>
Date: Tue, 10 Sep 2019 17:33:23 +0100
Subject: [PATCH] feat(static): add forEach() to KsqlStruct (MINOR)

Iterating across the values and their schemas is a common pattern, so lets add a method to facilitate.
---
 .../io/confluent/ksql/types/KsqlStruct.java   |  9 +++++
 .../confluent/ksql/types/KsqlStructTest.java  | 33 +++++++++++++++++++
 2 files changed, 42 insertions(+)

diff --git a/ksql-common/src/main/java/io/confluent/ksql/types/KsqlStruct.java b/ksql-common/src/main/java/io/confluent/ksql/types/KsqlStruct.java
index 856671899cc9..25e6e391fd37 100644
--- a/ksql-common/src/main/java/io/confluent/ksql/types/KsqlStruct.java
+++ b/ksql-common/src/main/java/io/confluent/ksql/types/KsqlStruct.java
@@ -25,6 +25,7 @@
 import java.util.List;
 import java.util.Objects;
 import java.util.Optional;
+import java.util.function.BiConsumer;
 
 /**
  * Instance of {@link io.confluent.ksql.schema.ksql.types.SqlStruct}.
@@ -55,6 +56,14 @@ public List<Optional<?>> values() {
     return values;
   }
 
+  public void forEach(final BiConsumer<? super Field, ? super Optional<?>> consumer) {
+    for (int idx = 0; idx < values.size(); idx++) {
+      final Field field = schema.fields().get(idx);
+      final Optional<?> value = values.get(idx);
+      consumer.accept(field, value);
+    }
+  }
+
   @Override
   public boolean equals(final Object o) {
     if (this == o) {
diff --git a/ksql-common/src/test/java/io/confluent/ksql/types/KsqlStructTest.java b/ksql-common/src/test/java/io/confluent/ksql/types/KsqlStructTest.java
index 5b21690f2e48..709c3ab88665 100644
--- a/ksql-common/src/test/java/io/confluent/ksql/types/KsqlStructTest.java
+++ b/ksql-common/src/test/java/io/confluent/ksql/types/KsqlStructTest.java
@@ -17,6 +17,7 @@
 
 import static org.hamcrest.MatcherAssert.assertThat;
 import static org.hamcrest.Matchers.contains;
+import static org.mockito.Mockito.inOrder;
 
 import io.confluent.ksql.schema.ksql.DataException;
 import io.confluent.ksql.schema.ksql.Field;
@@ -25,10 +26,16 @@
 import io.confluent.ksql.schema.ksql.types.SqlTypes;
 import io.confluent.ksql.util.KsqlException;
 import java.util.Optional;
+import java.util.function.BiConsumer;
 import org.junit.Rule;
 import org.junit.Test;
 import org.junit.rules.ExpectedException;
+import org.junit.runner.RunWith;
+import org.mockito.InOrder;
+import org.mockito.Mock;
+import org.mockito.junit.MockitoJUnitRunner;
 
+@RunWith(MockitoJUnitRunner.class)
 public class KsqlStructTest {
 
   private static final SqlStruct SCHEMA = SqlTypes.struct()
@@ -39,6 +46,9 @@ public class KsqlStructTest {
   @Rule
   public final ExpectedException expectedException = ExpectedException.none();
 
+  @Mock
+  private BiConsumer<? super Field, ? super Optional<?>> consumer;
+
   @Test
   public void shouldHandleExplicitNulls() {
     // When:
@@ -94,4 +104,27 @@ public void shouldBuildStruct() {
     // Then:
     assertThat(struct.values(), contains(Optional.of(10L), Optional.of(true)));
   }
+
+  @Test
+  public void shouldVisitFieldsInOrder() {
+    // Given:
+    final KsqlStruct struct = KsqlStruct.builder(SCHEMA)
+        .set(FieldName.of("f0"), Optional.of(10L))
+        .set(FieldName.of("s1", "v1"), Optional.of(true))
+        .build();
+
+    // When:
+    struct.forEach(consumer);
+
+    // Then:
+    final InOrder inOrder = inOrder(consumer);
+    inOrder.verify(consumer).accept(
+        struct.schema().fields().get(0),
+        struct.values().get(0)
+    );
+    inOrder.verify(consumer).accept(
+        struct.schema().fields().get(1),
+        struct.values().get(1)
+    );
+  }
 }
\ No newline at end of file