Text search for topic messages (#62)

* Text search for topic messages

* Code optimization

Author: apetrovs

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/ClusterService.java

@@ -147,9 +147,9 @@ private <T> Mono<T> updateCluster(T topic, String clusterName, KafkaCluster clus
                 });
     }
 
-    public Flux<TopicMessage> getMessages(String clusterName, String topicName, ConsumerPosition consumerPosition, Integer limit) {
+    public Flux<TopicMessage> getMessages(String clusterName, String topicName, ConsumerPosition consumerPosition, String query, Integer limit) {
         return clustersStorage.getClusterByName(clusterName)
-                .map(c -> consumingService.loadMessages(c, topicName, consumerPosition, limit))
+                .map(c -> consumingService.loadMessages(c, topicName, consumerPosition, query, limit))
                 .orElse(Flux.empty());
 
     }

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/service/ConsumingService.java

@@ -4,18 +4,22 @@
 import lombok.extern.log4j.Log4j2;
 
 import java.time.Duration;
+import java.util.LinkedList;
 import java.util.List;
 import java.util.Map;
 import java.util.Optional;
 import java.util.stream.Collectors;
 
+import org.apache.commons.lang3.StringUtils;
 import org.apache.kafka.clients.consumer.ConsumerRecord;
 import org.apache.kafka.clients.consumer.ConsumerRecords;
 import org.apache.kafka.clients.consumer.KafkaConsumer;
 import org.apache.kafka.common.TopicPartition;
 import org.apache.kafka.common.utils.Bytes;
 import org.springframework.stereotype.Service;
 
+import com.fasterxml.jackson.databind.JsonNode;
+import com.fasterxml.jackson.databind.ObjectMapper;
 import com.provectus.kafka.ui.cluster.deserialization.DeserializationService;
 import com.provectus.kafka.ui.cluster.deserialization.RecordDeserializer;
 import com.provectus.kafka.ui.cluster.model.ConsumerPosition;
@@ -36,12 +40,12 @@ public class ConsumingService {
 
 	private static final int MAX_RECORD_LIMIT = 100;
 	private static final int DEFAULT_RECORD_LIMIT = 20;
-	private static final int MAX_POLLS_COUNT = 30;
 
 	private final KafkaService kafkaService;
 	private final DeserializationService deserializationService;
+	private final ObjectMapper objectMapper = new ObjectMapper();
 
-	public Flux<TopicMessage> loadMessages(KafkaCluster cluster, String topic, ConsumerPosition consumerPosition, Integer limit) {
+	public Flux<TopicMessage> loadMessages(KafkaCluster cluster, String topic, ConsumerPosition consumerPosition, String query, Integer limit) {
 		int recordsLimit = Optional.ofNullable(limit)
 				.map(s -> Math.min(s, MAX_RECORD_LIMIT))
 				.orElse(DEFAULT_RECORD_LIMIT);
@@ -50,12 +54,44 @@ public Flux<TopicMessage> loadMessages(KafkaCluster cluster, String topic, Consu
 		return Flux.create(emitter::emit)
 				.subscribeOn(Schedulers.boundedElastic())
 				.map(r -> ClusterUtil.mapToTopicMessage(r, recordDeserializer))
+				.filter(m -> filterTopicMessage(m, query))
 				.limitRequest(recordsLimit);
 	}
 
+	private boolean filterTopicMessage(TopicMessage message, String query) {
+		if (StringUtils.isEmpty(query)) {
+			return true;
+		}
+
+		Object content = message.getContent();
+		JsonNode tree = objectMapper.valueToTree(content);
+		return treeContainsValue(tree, query);
+	}
+
+	private boolean treeContainsValue(JsonNode tree, String query) {
+		LinkedList<JsonNode> nodesForSearch = new LinkedList<>();
+		nodesForSearch.add(tree);
+
+		while (!nodesForSearch.isEmpty()) {
+			JsonNode node = nodesForSearch.removeFirst();
+
+			if (node.isContainerNode()) {
+				node.elements().forEachRemaining(nodesForSearch::add);
+				continue;
+			}
+
+			String nodeValue = node.asText();
+			if (nodeValue.contains(query)) {
+				return true;
+			}
+		}
+
+		return false;
+	}
+
 	@RequiredArgsConstructor
 	private static class RecordEmitter {
-
+		private static final int MAX_POLLS_COUNT = 30;
 		private static final Duration POLL_TIMEOUT_MS = Duration.ofMillis(1000L);
 
 		private final KafkaService kafkaService;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/MetricsRestController.java

@@ -65,9 +65,9 @@ public Mono<ResponseEntity<Flux<TopicConfig>>> getTopicConfigs(String clusterNam
     }
 
     @Override
-    public Mono<ResponseEntity<Flux<TopicMessage>>> getTopicMessages(String clusterName, String topicName, @Valid SeekType seekType, @Valid List<String> seekTo, @Valid Integer limit, ServerWebExchange exchange) {
+    public Mono<ResponseEntity<Flux<TopicMessage>>> getTopicMessages(String clusterName, String topicName, @Valid SeekType seekType, @Valid List<String> seekTo, @Valid Integer limit, @Valid String q, ServerWebExchange exchange) {
         return parseConsumerPosition(seekType, seekTo)
-                .map(consumerPosition -> ResponseEntity.ok(clusterService.getMessages(clusterName, topicName, consumerPosition, limit)));
+                .map(consumerPosition -> ResponseEntity.ok(clusterService.getMessages(clusterName, topicName, consumerPosition, q, limit)));
     }
 
     @Override

kafka-ui-contract/src/main/resources/swagger/kafka-ui-api.yaml

@@ -229,6 +229,10 @@ paths:
           in: query
           schema:
             type: integer
+        - name: q
+          in: query
+          schema:
+            type: string
       responses:
         200:
           description: OK