Merge pull request #102 from amosproj/feature/sprint-06-integration

sprint-06: integration

Author: jnsrnhld

README.md

@@ -12,18 +12,28 @@ We want to achieve this by structuring our UI according to the topology of Apach
 * Node.js Version **20.0.0** or higher
 * Docker Desktop
 
-## Backend
+## Backend and Pulsar instance
 
-First start Docker Desktop and create the pulsar setup from the root-directory with:
+First, build the application JAR from the `backend` directory with:
 
-```docker-compose up -d```
+```./mvnw package -DskipTests```
 
-Then start the application from the `backend` directory with:
 
-```./mvnw spring-boot:run```
+Then, start Docker Desktop and create the pulsar setup from the root-directory with:
+
+```docker-compose --profile demodata --profile backend up --build```
+
+Notes: 
+* `--build` is needed for the first startup, or when the demodata docker images are changed
+* `--profile demodata` is needed when you want to create the demo topology and start the demo producers & consumers, that will continuously send and receive messages
+* `--profile backend` is needed when you want to start the backend via the docker-compose setup. See below for starting it manually.
 
-After the start, all old topics are automatically removed from the Pulsar instance and some new topics, messages and
-consumers are added.
+### Starting the backend separately
+
+If you want to start the backend individually (e.g. during development), simply omit the `--profile backend` from the docker-compose command.
+Instead, start the application from the `backend` directory with:
+
+```./mvnw spring-boot:run```
 
 ### Tests
 

backend/Dockerfile

@@ -0,0 +1,3 @@
+FROM --platform=linux/amd64 openjdk:17-jdk-alpine
+COPY target/apachepulsarui-0.0.1-SNAPSHOT.jar apachepulsarui.jar
+ENTRYPOINT ["java","-jar","/apachepulsarui.jar"]

backend/pom.xml

@@ -95,12 +95,12 @@
 		<dependency>
 			<groupId>org.apache.pulsar</groupId>
 			<artifactId>pulsar-client</artifactId>
-			<version>2.11.1</version>
+			<version>3.0.0</version>
 		</dependency>
 		<dependency>
 			<groupId>org.apache.pulsar</groupId>
 			<artifactId>pulsar-client-admin</artifactId>
-			<version>2.11.1</version>
+			<version>3.0.0</version>
 		</dependency>
 
 	</dependencies>

backend/src/main/java/de/amos/apachepulsarui/ApachePulsarUiApplication.java

@@ -9,6 +9,9 @@
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cache.annotation.EnableCaching;
+import org.springframework.context.annotation.Bean;
+import org.springframework.web.servlet.config.annotation.CorsRegistry;
+import org.springframework.web.servlet.config.annotation.WebMvcConfigurer;
 
 @SpringBootApplication
 @EnableCaching
@@ -18,4 +21,14 @@ public static void main(String[] args) {
 		SpringApplication.run(ApachePulsarUiApplication.class, args);
 	}
 
+	@Bean
+	public WebMvcConfigurer configurer() {
+		return new WebMvcConfigurer() {
+			@Override
+			public void addCorsMappings(CorsRegistry corsRegistry) {
+				corsRegistry.addMapping("/**").allowedOrigins("*");
+			}
+		};
+	}
+
 }

backend/src/main/java/de/amos/apachepulsarui/controller/TopicController.java

@@ -6,8 +6,12 @@
 
 package de.amos.apachepulsarui.controller;
 
+import de.amos.apachepulsarui.dto.ProducerDto;
+import de.amos.apachepulsarui.dto.SubscriptionDto;
 import de.amos.apachepulsarui.dto.TopicDto;
 import de.amos.apachepulsarui.dto.TopicsDto;
+import de.amos.apachepulsarui.service.NamespaceService;
+import de.amos.apachepulsarui.service.TenantService;
 import de.amos.apachepulsarui.service.TopicService;
 import lombok.RequiredArgsConstructor;
 import org.apache.pulsar.common.naming.NamespaceName;
@@ -16,25 +20,51 @@
 import org.springframework.http.ResponseEntity;
 import org.springframework.web.bind.annotation.*;
 
+import java.util.List;
+
 
 @RestController
 @RequiredArgsConstructor
 @RequestMapping("/topic")
 public class TopicController {
 
+    private final TenantService tenantService;
+    private final NamespaceService namespaceService;
     private final TopicService topicService;
 
+    // Talked about this with Julian - probably we won't use it this way later, but at first it's easier for them
+    // to just get all topics at once.
+    @GetMapping("/all")
+    public ResponseEntity<TopicsDto> getAll() {
+        List<TopicDto> allTopics = tenantService.getAllTenants().stream()
+                .flatMap(tenantDto -> namespaceService.getAllOfTenant(tenantDto).stream())
+                .flatMap(namespaceDto -> topicService.getAllByNamespace(namespaceDto.getId()).stream())
+                .toList();
+        return new ResponseEntity<>(new TopicsDto(allTopics), HttpStatus.OK);
+    }
+
     @GetMapping("/{tenant}/{namespace}")
     public ResponseEntity<TopicsDto> getTopicsByNamespace(@PathVariable String namespace, @PathVariable String tenant) {
         String namespaceName = NamespaceName.get(tenant, namespace).toString();
         return new ResponseEntity<>(new TopicsDto(topicService.getAllByNamespace(namespaceName)), HttpStatus.OK);
     }
 
-    @GetMapping("/{topic}")
-    public ResponseEntity<TopicDto> getTopicWithMessagesByName(@PathVariable String topic) {
-       return new ResponseEntity<>(topicService.getTopicWithMessagesByName(topic), HttpStatus.OK);
+    @GetMapping
+    public ResponseEntity<TopicDto> getTopicWithMessagesByName(@RequestParam String name) {
+       return new ResponseEntity<>(topicService.getTopicWithMessagesByName(name), HttpStatus.OK);
+    }
+
+    @GetMapping("/subscription/{subscription}")
+    public ResponseEntity<SubscriptionDto> getSubyscriptionByNameAndTopic(@RequestParam String topic, @PathVariable String subscription) {
+        return new ResponseEntity<>(topicService.getSubscriptionByTopic(topic, subscription), HttpStatus.OK);
+    }
+
+    @GetMapping("/producer/{producer}")
+    public ResponseEntity<ProducerDto> getProducerByNameAndTopic(@RequestParam String topic, @PathVariable String producer) {
+        return new ResponseEntity<>(topicService.getProducerByTopic(topic, producer), HttpStatus.OK);
     }
 
+
     @PostMapping("/new")
     public ResponseEntity<Void> newTopic(@RequestParam String topic) {
         if (!TopicName.isValid(topic)) {

backend/src/main/java/de/amos/apachepulsarui/dto/ConsumerDto.java

@@ -16,7 +16,7 @@ public class ConsumerDto {
 
     private String name;
 
-    public static ConsumerDto createConsumerDto(ConsumerStats consumerStats) {
+    public static ConsumerDto create(ConsumerStats consumerStats) {
         return ConsumerDto.builder().name(consumerStats.getConsumerName()).build();
     }
 

backend/src/main/java/de/amos/apachepulsarui/dto/MessageDto.java

@@ -35,6 +35,8 @@ public class MessageDto {
 
     Long publishTime;
 
+    String producer;
+
     /**
      * Static factory for messages already existing in Pulsar.
      */
@@ -48,6 +50,7 @@ public static MessageDto fromExistingMessage(Message<byte[]> message, String sch
         messageDto.namespace = TopicName.get(topicName).getNamespacePortion();
         messageDto.tenant = TopicName.get(topicName).getTenant();
         messageDto.publishTime = message.getPublishTime();
+        messageDto.producer = message.getProducerName();
 
         return messageDto;
     }

backend/src/main/java/de/amos/apachepulsarui/dto/ProducerDto.java

@@ -5,18 +5,27 @@
 
 package de.amos.apachepulsarui.dto;
 
-import lombok.AccessLevel;
-import lombok.Builder;
-import lombok.Data;
+import lombok.*;
 import org.apache.pulsar.common.policies.data.PublisherStats;
 
+import java.util.List;
+
 @Data
+@AllArgsConstructor
 @Builder(access = AccessLevel.PRIVATE)
 public class ProducerDto {
 
     private String name;
 
-    public static ProducerDto createProducerDto(PublisherStats publisherStats) {
-        return ProducerDto.builder().name(publisherStats.getProducerName()).build();
+    private List<MessageDto> messagesDto;
+
+    private long amountOfMessages;
+
+    public static ProducerDto create(PublisherStats publisherStats, List<MessageDto> messages) {
+        return ProducerDto.builder()
+                .name(publisherStats.getProducerName())
+                .messagesDto(messages)
+                .amountOfMessages(messages.size())
+                .build();
     }
 }

backend/src/main/java/de/amos/apachepulsarui/dto/SubscriptionDto.java

@@ -8,24 +8,55 @@
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Data;
+import org.apache.pulsar.common.policies.data.SubscriptionStats;
 
 import java.util.List;
+import java.util.Objects;
 
 @Data
 @Builder(access = AccessLevel.PRIVATE)
 public class SubscriptionDto {
 
     private String name;
 
-    private List<ConsumerDto> consumers;
+    private ConsumerDto activeConsumer;
+
+    private List<ConsumerDto> allConsumers;
+
+    private List<MessageDto> messages;
 
     private long numberConsumers;
 
-    public static SubscriptionDto createSubscriptionDto(String name, List<ConsumerDto> consumers) {
+    private double msgAckRate;
+
+    private long msgBacklog;
+
+    private long msgOutCounter;
+
+    public static SubscriptionDto create(SubscriptionStats subscriptionStats, List<MessageDto> messages, String name) {
+        List<ConsumerDto> consumers = getConsumers(subscriptionStats);
+        ConsumerDto active = consumers
+                .stream()
+                .filter(c -> Objects.equals(c.getName(), subscriptionStats.getActiveConsumerName()))
+                .findFirst()
+                .orElse(null);
+
         return SubscriptionDto.builder()
                 .name(name)
-                .consumers(consumers)
-                .numberConsumers(consumers.size())
+                .messages(messages)
+                .activeConsumer(active)
+                .allConsumers(consumers)
+                .msgAckRate(subscriptionStats.getMessageAckRate())
+                .msgBacklog(subscriptionStats.getMsgBacklog())
+                .msgOutCounter(builder().msgOutCounter)
                 .build();
     }
+
+    private static List<ConsumerDto> getConsumers(SubscriptionStats subscriptionStats) {
+        return subscriptionStats.getConsumers()
+                .stream()
+                .map(ConsumerDto::create)
+                .toList();
+
+    }
 }

backend/src/main/java/de/amos/apachepulsarui/dto/TopicStatsDto.java

@@ -8,21 +8,18 @@
 import lombok.AccessLevel;
 import lombok.Builder;
 import lombok.Data;
-import org.apache.pulsar.common.policies.data.ConsumerStats;
 import org.apache.pulsar.common.policies.data.PublisherStats;
-import org.apache.pulsar.common.policies.data.SubscriptionStats;
 import org.apache.pulsar.common.policies.data.TopicStats;
 
 import java.util.ArrayList;
-import java.util.HashMap;
 import java.util.List;
-import java.util.Map;
+import java.util.Set;
 
 @Data
 @Builder(access = AccessLevel.PRIVATE)
 public class TopicStatsDto {
 
-    private List<String> subscriptions;
+    private Set<String> subscriptions;
 
     private List<String> producers;
 
@@ -39,7 +36,8 @@ public class TopicStatsDto {
     private long storageSize;
 
     public static TopicStatsDto createTopicStatsDto(TopicStats topicStats) {
-        List<String> subscriptions = getSubscriptions(topicStats);
+
+        Set<String> subscriptions = topicStats.getSubscriptions().keySet();
         List<String> producers = getProducers(topicStats);
 
        return TopicStatsDto.builder()
@@ -55,27 +53,16 @@ public static TopicStatsDto createTopicStatsDto(TopicStats topicStats) {
 
     }
 
-    private static List<String> getProducers(TopicStats topicStats) {
+    public static List<String> getProducers(TopicStats topicStats) {
         List<PublisherStats> publisherStats = new ArrayList<>(topicStats.getPublishers());
         return publisherStats.stream()
                 .map(PublisherStats::getProducerName)
                 .toList();
     }
 
-    private static List<String> getSubscriptions(TopicStats topicStats) {
-        Map<String, SubscriptionStats> subscriptionStats = new HashMap<>(topicStats.getSubscriptions());
-        List<String> subscriptions = new ArrayList<>();
-        subscriptionStats.forEach((name, producers) -> subscriptions.add(name));
-        return subscriptions;
-    }
 
-    private static List<ConsumerDto> getConsumers(SubscriptionStats subscriptionStats) {
-        List<ConsumerStats> consumerStats = new ArrayList<>(subscriptionStats.getConsumers());
-        return consumerStats
-                .stream()
-                .map(ConsumerDto::createConsumerDto)
-                .toList();
-    }
+
+
 
 
 }