BE: Messages: Allow searching in message headers (#695)

Author: yeikel

api/src/main/java/io/kafbat/ui/emitter/MessageFilters.java

@@ -52,7 +52,23 @@ public static Predicate<TopicMessageDTO> noop() {
 
   public static Predicate<TopicMessageDTO> containsStringFilter(String string) {
     return msg -> StringUtils.contains(msg.getKey(), string)
-        || StringUtils.contains(msg.getContent(), string);
+        || StringUtils.contains(msg.getContent(), string) || headersContains(msg, string);
+  }
+
+  private static boolean headersContains(TopicMessageDTO msg, String searchString) {
+    final var headers = msg.getHeaders();
+
+    if (headers == null) {
+      return false;
+    }
+
+    for (final var entry : headers.entrySet()) {
+      if (StringUtils.contains(entry.getKey(), searchString) || StringUtils.contains(entry.getValue(), searchString)) {
+        return true;
+      }
+    }
+
+    return false;
   }
 
   public static Predicate<TopicMessageDTO> celScriptFilter(String script) {

api/src/test/java/io/kafbat/ui/emitter/MessageFiltersTest.java

@@ -28,7 +28,7 @@ class StringContainsFilter {
     Predicate<TopicMessageDTO> filter = containsStringFilter("abC");
 
     @Test
-    void returnsTrueWhenStringContainedInKeyOrContentOrInBoth() {
+    void returnsTrueWhenStringContainedInKeyOrContentOrHeadersOrInAllThree() {
       assertTrue(
           filter.test(msg().key("contains abCd").content("some str"))
       );
@@ -40,6 +40,14 @@ void returnsTrueWhenStringContainedInKeyOrContentOrInBoth() {
       assertTrue(
           filter.test(msg().key("contains abCd").content("contains abCd"))
       );
+
+      assertTrue(
+          filter.test(msg().key("dfg").content("does-not-contain").headers(Map.of("abC", "value")))
+      );
+
+      assertTrue(
+          filter.test(msg().key("dfg").content("does-not-contain").headers(Map.of("x1", "some abC")))
+      );
     }
 
     @Test
@@ -55,6 +63,11 @@ void returnsFalseOtherwise() {
       assertFalse(
           filter.test(msg().key("aBc").content("AbC"))
       );
+
+      assertFalse(
+          filter.test(msg().key("aBc").content("AbC").headers(Map.of("abc", "value")))
+      );
+
     }
 
   }