Hw13

dev/docker-compose.yml

@@ -20,8 +20,11 @@ services:
       - PG_DB_URL=jdbc:postgresql://postgres:5432/postgres?currentSchema=public&stringtype=unspecified
       - PG_DB_USERNAME=postgres
       - PG_DB_PASSWORD=123
+      - SPRING_KAFKA_BOOTSTRAP-SERVERS=kafka:9092
+      - SPRING_KAFKA_CONSUMER_GROUP-ID=consumers
     depends_on:
       - mongo
+      - kafka
 
   landscape-service:
     container_name: landscape-service
@@ -38,10 +41,13 @@ services:
       - DB_PASSWORD=123
       - RANCHER_REST_URL=http://rancher-service:8082
       - HANDYMAN_REST_URL=http://handyman-service:8080
+      - SPRING_KAFKA_BOOTSTRAP-SERVERS=kafka:9092
+      - SPRING_KAFKA_CONSUMER_GROUP-ID=consumers
     depends_on:
       - rancher-service
       - handyman-service
       - postgres
+      - kafka
 
   rancher-service:
     container_name: rancher-service
@@ -58,8 +64,11 @@ services:
       - PG_DB_URL=jdbc:postgresql://postgres:5432/postgres?currentSchema=public&stringtype=unspecified
       - PG_DB_USERNAME=postgres
       - PG_DB_PASSWORD=123
+      - SPRING_KAFKA_BOOTSTRAP-SERVERS=kafka:9092
+      - SPRING_KAFKA_CONSUMER_GROUP-ID=consumers
     depends_on:
       - mongo
+      - kafka
 
   postgres:
     image: postgis/postgis:15-3.3
@@ -127,3 +136,47 @@ services:
       - app-tier
     depends_on:
       - prometheus
+
+  zookeeper:
+    container_name: zookeeper
+    image: "bitnami/zookeeper:latest"
+    networks:
+      - app-tier
+    ports:
+      - "2181:2181"
+    environment:
+      ALLOW_ANONYMOUS_LOGIN: "yes"
+
+  kafka:
+    container_name: kafka
+    image: "bitnami/kafka:3.0.0"
+    networks:
+      - app-tier
+    ports:
+      - "9093:9093"
+      - "9092:9092"
+    environment:
+      - KAFKA_BROKER_ID=1
+      - KAFKA_CFG_LISTENER_SECURITY_PROTOCOL_MAP=CLIENT:PLAINTEXT,EXTERNAL:PLAINTEXT
+      - KAFKA_CFG_LISTENERS=CLIENT://:9092,EXTERNAL://:9093
+      - KAFKA_CFG_ADVERTISED_LISTENERS=CLIENT://kafka:9092,EXTERNAL://localhost:9093
+      - KAFKA_CFG_INTER_BROKER_LISTENER_NAME=CLIENT
+      - KAFKA_CFG_ZOOKEEPER_CONNECT=zookeeper:2181
+      - KAFKA_CFG_AUTO_CREATE_TOPICS_ENABLE=true
+      - ALLOW_PLAINTEXT_LISTENER=yes
+    depends_on:
+      - zookeeper
+
+  kafka-ui:
+    container_name: kafka-ui
+    image: "provectuslabs/kafka-ui"
+    networks:
+      - app-tier
+    ports:
+      - "8090:8080"
+    environment:
+      - KAFKA_CLUSTERS_0_NAME=kafka
+      - KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS=kafka:9092
+    restart: unless-stopped
+    depends_on:
+      - kafka

handyman-service/build.gradle

@@ -64,6 +64,7 @@ ext {
 }
 
 dependencies {
+    implementation 'org.springframework.kafka:spring-kafka'
     implementation 'org.n52.jackson:jackson-datatype-jts:1.2.10'
     implementation 'org.locationtech.jts:jts-io:1.18.2'
     implementation 'org.locationtech.jts:jts-core:1.18.2'

handyman-service/src/main/java/dev/archie/handymanservice/handyman/HandymanController.java

@@ -1,8 +1,9 @@
 package dev.archie.handymanservice.handyman;
 
 import dev.archie.handymanservice.handyman.dto.CreatingHandymanDto;
-import dev.archie.handymanservice.landscape.Order;
+import dev.archie.landscapeservice.order.dto.SendOrderToUser;
 import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.annotation.KafkaListener;
 import org.springframework.web.bind.annotation.DeleteMapping;
 import org.springframework.web.bind.annotation.GetMapping;
 import org.springframework.web.bind.annotation.PathVariable;
@@ -12,8 +13,6 @@
 import org.springframework.web.bind.annotation.RequestMapping;
 import org.springframework.web.bind.annotation.RestController;
 
-import java.util.UUID;
-
 @RestController
 @RequiredArgsConstructor
 @RequestMapping("/handymen")
@@ -57,8 +56,8 @@ public void delete(@PathVariable String id) {
         handymanService.delete(id);
     }
 
-    @PostMapping("/order/{innerId}")
-    public boolean orderAJob(@PathVariable UUID innerId, @RequestBody Order order) {
-        return handymanService.orderAJob(innerId, order);
+    @KafkaListener(topics = "handyman.notify.order")
+    public void orderAJob(SendOrderToUser orderToUser) {
+        handymanService.orderAJob(orderToUser.handyman().getId(), orderToUser.order());
     }
 }

handyman-service/src/main/java/dev/archie/handymanservice/handyman/HandymanService.java

@@ -9,12 +9,12 @@
 import dev.archie.handymanservice.landscape.HandymanClient;
 import dev.archie.handymanservice.landscape.LandscapeService;
 import dev.archie.handymanservice.landscape.Order;
-import dev.archie.handymanservice.landscape.OrderClient;
 import dev.archie.handymanservice.landscape.WorkStatus;
-import dev.archie.handymanservice.landscape.dto.CreatingOrderDto;
 import dev.archie.handymanservice.landscape.dto.HandymanDto;
+import dev.archie.landscapeservice.order.dto.SendOrderToUser;
 import lombok.RequiredArgsConstructor;
 import lombok.extern.slf4j.Slf4j;
+import org.springframework.kafka.core.KafkaTemplate;
 import org.springframework.stereotype.Service;
 
 import java.util.List;
@@ -36,7 +36,7 @@ public class HandymanService {
     private final AccountService accountService;
 
     private final HandymanClient handymanClient;
-    private final OrderClient orderClient;
+    private final KafkaTemplate<String, Object> kafkaTemplate;
 
     /**
      * @param creatingHandymanDto handyman dto create. Email should not exist
@@ -125,8 +125,9 @@ private Handyman mapUserToHandyman(HandymanDto createdUser, CreatingHandymanDto
                 .build();
     }
 
-    public boolean orderAJob(UUID innerId, Order order) {
+    public void orderAJob(UUID innerId, Order order) {
         if (!handymanRepository.existsByInnerId(innerId)) {
+            kafkaTemplate.send("landscape.order.cancel", order);
             throw new NoSuchUserException(innerId.toString());
         }
         boolean acceptedTheJob = order.getSkills().stream()
@@ -136,8 +137,8 @@ public boolean orderAJob(UUID innerId, Order order) {
             log.info(innerId + " принял работу");
         } else {
             log.info(innerId + " отказался от работы");
+            kafkaTemplate.send("landscape.order.cancel", order);
         }
-        return acceptedTheJob;
     }
 
     public void getTheJobDone(Order order, UUID innerId) {
@@ -148,14 +149,7 @@ public void getTheJobDone(Order order, UUID innerId) {
             throw new RuntimeException("Execution was interrupted" + e);
         }
         order.setStatus(WorkStatus.DONE);
-        CreatingOrderDto updatingDto = CreatingOrderDto.builder()
-                .status(order.getStatus())
-                .fieldId(order.getField().getId())
-                .userId(innerId)
-                .workType(order.getWorkType())
-                .skills(order.getSkills().stream().map(Skill::getName).toList())
-                .build();
-        orderClient.update(order.getId(), updatingDto);
+        kafkaTemplate.send("landscape.order.complete", new SendOrderToUser(null, order));
         log.info(innerId + " выполнил работу");
     }
 

handyman-service/src/main/java/dev/archie/handymanservice/landscape/User.java

@@ -1,12 +1,16 @@
 package dev.archie.handymanservice.landscape;
 
+import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Data;
+import lombok.NoArgsConstructor;
 
 import java.time.LocalDateTime;
 import java.util.UUID;
 
 @Data
+@AllArgsConstructor
+@NoArgsConstructor
 @Builder
 public class User {
 

handyman-service/src/main/java/dev/archie/landscapeservice/order/dto/SendOrderToUser.java

@@ -0,0 +1,10 @@
+package dev.archie.landscapeservice.order.dto;
+
+import dev.archie.handymanservice.landscape.Order;
+import dev.archie.handymanservice.landscape.User;
+
+public record SendOrderToUser(
+        User handyman,
+        Order order
+) {
+}

handyman-service/src/main/resources/application.properties

@@ -14,4 +14,16 @@ spring.datasource.url=${PG_DB_URL}
 spring.datasource.username=${PG_DB_USERNAME}
 spring.datasource.password=${PG_DB_PASSWORD}
 
-application.landscape.rest.url=${LANDSCAPE_REST_URL}
+application.landscape.rest.url=${LANDSCAPE_REST_URL}
+
+spring.kafka.bootstrap-servers=${SPRING_KAFKA_BOOTSTRAP-SERVERS}
+
+spring.kafka.consumer.group-id=${SPRING_KAFKA_CONSUMER_GROUP-ID}
+spring.kafka.consumer.auto-offset-reset=earliest
+spring.kafka.consumer.enable-auto-commit=true
+spring.kafka.consumer.key-deserializer=org.apache.kafka.common.serialization.StringDeserializer
+spring.kafka.consumer.value-deserializer=org.springframework.kafka.support.serializer.JsonDeserializer
+spring.kafka.consumer.properties.spring.json.trusted.packages=*
+
+spring.kafka.producer.key-serializer=org.apache.kafka.common.serialization.StringSerializer
+spring.kafka.producer.value-serializer=org.springframework.kafka.support.serializer.JsonSerializer

landscape-service/build.gradle

@@ -53,6 +53,7 @@ jar {
 }
 
 dependencies {
+    implementation 'org.springframework.kafka:spring-kafka'
     implementation 'org.springframework.cloud:spring-cloud-starter-openfeign'
 
     implementation 'org.hibernate:hibernate-spatial'

landscape-service/src/main/java/dev/archie/landscapeservice/field/Field.java

@@ -1,17 +1,14 @@
 package dev.archie.landscapeservice.field;
 
+import com.fasterxml.jackson.annotation.JsonIgnore;
 import com.fasterxml.jackson.annotation.JsonIgnoreProperties;
-import com.fasterxml.jackson.databind.annotation.JsonDeserialize;
-import com.fasterxml.jackson.databind.annotation.JsonSerialize;
 import dev.archie.landscapeservice.gardener.Gardener;
 import lombok.AllArgsConstructor;
 import lombok.Builder;
 import lombok.Getter;
 import lombok.NoArgsConstructor;
 import lombok.Setter;
 import org.locationtech.jts.geom.Geometry;
-import org.n52.jackson.datatype.jts.GeometryDeserializer;
-import org.n52.jackson.datatype.jts.GeometrySerializer;
 
 import javax.persistence.Column;
 import javax.persistence.Entity;
@@ -44,11 +41,10 @@ public class Field {
     @Column(name = "longitude", nullable = false)
     private Double longitude;
 
-    @JsonSerialize(using = GeometrySerializer.class)
-    @JsonDeserialize(contentUsing = GeometryDeserializer.class)
+    @JsonIgnore
     private Geometry area;
 
-    @JsonIgnoreProperties("fields")
+    @JsonIgnoreProperties({"fields", "orders"})
     @ManyToOne
     @JoinColumn(name = "gardener_id")
     private Gardener gardener;

landscape-service/src/main/java/dev/archie/landscapeservice/handyman/HandymanRepository.java

@@ -11,7 +11,6 @@ public interface HandymanRepository extends JpaRepository<Handyman, UUID> {
             select h
             from Handyman h
             join User u on u.id = h.id
-            where u.orders.size = 0
             order by ST_Distance(ST_Point(?1, ?2), ST_Point(u.latitude, u.longitude))
             """)
     List<Handyman> findNearestNotBusyHandyman(Double latitude, Double longitude);

landscape-service/src/main/java/dev/archie/landscapeservice/order/Order.java

@@ -15,6 +15,7 @@
 import javax.persistence.Entity;
 import javax.persistence.EnumType;
 import javax.persistence.Enumerated;
+import javax.persistence.FetchType;
 import javax.persistence.GeneratedValue;
 import javax.persistence.GenerationType;
 import javax.persistence.Id;
@@ -24,7 +25,6 @@
 import javax.persistence.OneToOne;
 import javax.persistence.Table;
 import java.time.LocalDateTime;
-import java.util.LinkedHashSet;
 import java.util.Set;
 
 @Builder
@@ -55,15 +55,15 @@ public class Order {
     @JoinColumn(name = "field_id")
     private Field field;
 
-
     @JsonIgnoreProperties("orders")
     @ManyToOne
     @JoinColumn(name = "user_id")
     private User user;
 
-    @OneToMany
+    @JsonIgnoreProperties("handyman")
+    @OneToMany(fetch = FetchType.EAGER)
     @JoinColumn(name = "order_id")
-    private Set<Skill> skills = new LinkedHashSet<>();
+    private Set<Skill> skills;
 
     @Column(name = "grade", nullable = false)
     private int grade;