Feature refactoring, structure simplifying (provectus#250)

kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/KafkaConnectClients.java

@@ -0,0 +1,15 @@
+package com.provectus.kafka.ui.client;
+
+import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+public final class KafkaConnectClients {
+
+    private static final Map<String, KafkaConnectClientApi> CACHE = new ConcurrentHashMap<>();
+
+    public static KafkaConnectClientApi withBaseUrl(String basePath) {
+        return CACHE.computeIfAbsent(basePath, RetryingKafkaConnectClient::new);
+    }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/client/RetryingKafkaConnectClient.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/client/RetryingKafkaConnectClient.java

@@ -1,9 +1,9 @@
-package com.provectus.kafka.ui.cluster.client;
+package com.provectus.kafka.ui.client;
 
-import com.provectus.kafka.ui.cluster.exception.RebalanceInProgressException;
-import com.provectus.kafka.ui.cluster.exception.ValidationException;
+import com.provectus.kafka.ui.exception.RebalanceInProgressException;
+import com.provectus.kafka.ui.exception.ValidationException;
 import com.provectus.kafka.ui.connect.ApiClient;
-import com.provectus.kafka.ui.connect.api.ConnectApi;
+import com.provectus.kafka.ui.connect.api.KafkaConnectClientApi;
 import com.provectus.kafka.ui.connect.model.Connector;
 import com.provectus.kafka.ui.connect.model.NewConnector;
 import lombok.extern.log4j.Log4j2;
@@ -22,7 +22,7 @@
 import java.util.Map;
 
 @Log4j2
-public class RetryingKafkaConnectClient extends ConnectApi {
+public class RetryingKafkaConnectClient extends KafkaConnectClientApi {
     private static final int MAX_RETRIES = 5;
 
     public RetryingKafkaConnectClient(String basePath) {

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/config/ClustersProperties.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ClustersProperties.java

@@ -1,4 +1,4 @@
-package com.provectus.kafka.ui.cluster.config;
+package com.provectus.kafka.ui.config;
 
 import java.util.Properties;
 import lombok.Data;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/config/Config.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/Config.java

@@ -1,6 +1,6 @@
-package com.provectus.kafka.ui.cluster.config;
+package com.provectus.kafka.ui.config;
 
-import com.provectus.kafka.ui.cluster.util.JmxPoolFactory;
+import com.provectus.kafka.ui.util.JmxPoolFactory;
 import org.apache.commons.pool2.KeyedObjectPool;
 import org.apache.commons.pool2.impl.GenericKeyedObjectPool;
 import org.apache.commons.pool2.impl.GenericKeyedObjectPoolConfig;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/config/CorsGlobalConfiguration.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/CorsGlobalConfiguration.java

@@ -1,4 +1,4 @@
-package com.provectus.kafka.ui.rest.config;
+package com.provectus.kafka.ui.config;
 
 import org.springframework.context.annotation.Configuration;
 import org.springframework.context.annotation.Profile;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/config/CustomWebFilter.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/CustomWebFilter.java

@@ -1,4 +1,4 @@
-package com.provectus.kafka.ui.rest.config;
+package com.provectus.kafka.ui.config;
 
 import org.springframework.stereotype.Component;
 import org.springframework.web.server.ServerWebExchange;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/config/ReadOnlyModeFilter.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/ReadOnlyModeFilter.java

@@ -1,8 +1,8 @@
-package com.provectus.kafka.ui.rest.config;
+package com.provectus.kafka.ui.config;
 
-import com.provectus.kafka.ui.cluster.exception.NotFoundException;
-import com.provectus.kafka.ui.cluster.exception.ReadOnlyException;
-import com.provectus.kafka.ui.cluster.model.ClustersStorage;
+import com.provectus.kafka.ui.exception.NotFoundException;
+import com.provectus.kafka.ui.exception.ReadOnlyException;
+import com.provectus.kafka.ui.service.ClustersStorage;
 import lombok.RequiredArgsConstructor;
 import org.jetbrains.annotations.NotNull;
 import org.springframework.core.annotation.Order;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/cluster/config/SecurityConfig.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/config/SecurityConfig.java

@@ -1,4 +1,4 @@
-package com.provectus.kafka.ui.cluster.config;
+package com.provectus.kafka.ui.config;
 
 import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
 import org.springframework.context.annotation.Bean;

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/BrokersController.java

@@ -0,0 +1,32 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.BrokersApi;
+import com.provectus.kafka.ui.service.ClusterService;
+import com.provectus.kafka.ui.model.Broker;
+import com.provectus.kafka.ui.model.BrokerMetrics;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+@Log4j2
+public class BrokersController implements BrokersApi {
+  private final ClusterService clusterService;
+
+  @Override
+  public Mono<ResponseEntity<BrokerMetrics>> getBrokersMetrics(String clusterName, Integer id, ServerWebExchange exchange) {
+    return clusterService.getBrokerMetrics(clusterName, id)
+        .map(ResponseEntity::ok)
+        .onErrorReturn(ResponseEntity.notFound().build());
+  }
+
+  @Override
+  public Mono<ResponseEntity<Flux<Broker>>> getBrokers(String clusterName, ServerWebExchange exchange) {
+    return Mono.just(ResponseEntity.ok(clusterService.getBrokers(clusterName)));
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/ClustersController.java

@@ -0,0 +1,40 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.ClustersApi;
+import com.provectus.kafka.ui.service.ClusterService;
+import com.provectus.kafka.ui.model.Cluster;
+import com.provectus.kafka.ui.model.ClusterMetrics;
+import com.provectus.kafka.ui.model.ClusterStats;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+@Log4j2
+public class ClustersController implements ClustersApi {
+  private final ClusterService clusterService;
+
+  @Override
+  public Mono<ResponseEntity<ClusterMetrics>> getClusterMetrics(String clusterName, ServerWebExchange exchange) {
+    return clusterService.getClusterMetrics(clusterName)
+        .map(ResponseEntity::ok)
+        .onErrorReturn(ResponseEntity.notFound().build());
+  }
+
+  @Override
+  public Mono<ResponseEntity<ClusterStats>> getClusterStats(String clusterName, ServerWebExchange exchange) {
+    return clusterService.getClusterStats(clusterName)
+        .map(ResponseEntity::ok)
+        .onErrorReturn(ResponseEntity.notFound().build());
+  }
+
+  @Override
+  public Mono<ResponseEntity<Flux<Cluster>>> getClusters(ServerWebExchange exchange) {
+    return Mono.just(ResponseEntity.ok(Flux.fromIterable(clusterService.getClusters())));
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/ConsumerGroupsController.java

@@ -0,0 +1,35 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.ConsumerGroupsApi;
+import com.provectus.kafka.ui.service.ClusterService;
+import com.provectus.kafka.ui.model.ConsumerGroup;
+import com.provectus.kafka.ui.model.ConsumerGroupDetails;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+@Log4j2
+public class ConsumerGroupsController implements ConsumerGroupsApi {
+  private final ClusterService clusterService;
+
+  @Override
+  public Mono<ResponseEntity<ConsumerGroupDetails>> getConsumerGroup(
+      String clusterName, String consumerGroupId, ServerWebExchange exchange) {
+    return clusterService.getConsumerGroupDetail(clusterName, consumerGroupId).map(ResponseEntity::ok);
+  }
+
+
+  @Override
+  public Mono<ResponseEntity<Flux<ConsumerGroup>>> getConsumerGroups(String clusterName, ServerWebExchange exchange) {
+    return clusterService.getConsumerGroups(clusterName)
+        .map(Flux::fromIterable)
+        .map(ResponseEntity::ok)
+        .switchIfEmpty(Mono.just(ResponseEntity.notFound().build())); // TODO: check behaviour on cluster not found and empty groups list
+  }
+}

kafka-ui-api/src/main/java/com/provectus/kafka/ui/rest/KafkaConnectRestController.java → kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/KafkaConnectController.java

@@ -1,7 +1,7 @@
-package com.provectus.kafka.ui.rest;
+package com.provectus.kafka.ui.controller;
 
-import com.provectus.kafka.ui.api.ApiClustersConnectsApi;
-import com.provectus.kafka.ui.cluster.service.KafkaConnectService;
+import com.provectus.kafka.ui.api.KafkaConnectApi;
+import com.provectus.kafka.ui.service.KafkaConnectService;
 import com.provectus.kafka.ui.model.*;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.log4j.Log4j2;
@@ -17,7 +17,7 @@
 @RestController
 @RequiredArgsConstructor
 @Log4j2
-public class KafkaConnectRestController implements ApiClustersConnectsApi {
+public class KafkaConnectController implements KafkaConnectApi {
     private final KafkaConnectService kafkaConnectService;
 
     @Override

kafka-ui-api/src/main/java/com/provectus/kafka/ui/controller/MessagesController.java

@@ -0,0 +1,63 @@
+package com.provectus.kafka.ui.controller;
+
+import com.provectus.kafka.ui.api.MessagesApi;
+import com.provectus.kafka.ui.model.ConsumerPosition;
+import com.provectus.kafka.ui.service.ClusterService;
+import com.provectus.kafka.ui.model.SeekType;
+import com.provectus.kafka.ui.model.TopicMessage;
+import java.util.Collections;
+import java.util.List;
+import java.util.Optional;
+import java.util.function.Function;
+import javax.validation.Valid;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.log4j.Log4j2;
+import org.apache.commons.lang3.tuple.Pair;
+import org.springframework.http.ResponseEntity;
+import org.springframework.web.bind.annotation.RestController;
+import org.springframework.web.server.ServerWebExchange;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+
+@RestController
+@RequiredArgsConstructor
+@Log4j2
+public class MessagesController implements MessagesApi {
+  private final ClusterService clusterService;
+
+  @Override
+  public Mono<ResponseEntity<Void>> deleteTopicMessages(
+      String clusterName, String topicName, @Valid List<Integer> partitions, ServerWebExchange exchange) {
+    return clusterService.deleteTopicMessages(
+        clusterName,
+        topicName,
+        Optional.ofNullable(partitions).orElse(List.of())
+    ).map(ResponseEntity::ok);
+  }
+
+
+  @Override
+  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, q, limit)));
+  }
+
+  private Mono<ConsumerPosition> parseConsumerPosition(SeekType seekType, List<String> seekTo) {
+    return Mono.justOrEmpty(seekTo)
+        .defaultIfEmpty(Collections.emptyList())
+        .flatMapIterable(Function.identity())
+        .map(p -> {
+          String[] splited = p.split("::");
+          if (splited.length != 2) {
+            throw new IllegalArgumentException("Wrong seekTo argument format. See API docs for details");
+          }
+
+          return Pair.of(Integer.parseInt(splited[0]), Long.parseLong(splited[1]));
+        })
+        .collectMap(Pair::getKey, Pair::getValue)
+        .map(positions -> new ConsumerPosition(seekType != null ? seekType : SeekType.BEGINNING, positions));
+  }
+
+}