Add messaging template

src/main/java/ru/nsu/fit/directors/userservice/configuration/WebSocketConfiguration.java

@@ -13,15 +13,15 @@ public class WebSocketConfiguration implements WebSocketMessageBrokerConfigurer
     @Override
     public void configureMessageBroker(MessageBrokerRegistry config) {
         config.enableSimpleBroker("/topic");
-        config.setApplicationDestinationPrefixes("/business");
+        config.setApplicationDestinationPrefixes("/user");
     }
 
     @Override
     public void registerStompEndpoints(StompEndpointRegistry registry) {
-        registry.addEndpoint("/business/chat")
+        registry.addEndpoint("/user/chat")
             .setAllowedOriginPatterns("*")
             .withSockJS();
-        registry.addEndpoint("/business/chat")
+        registry.addEndpoint("/user/chat")
             .setAllowedOriginPatterns("*");
     }
 }

src/main/java/ru/nsu/fit/directors/userservice/event/BusinessMessageEvent.java

@@ -0,0 +1,4 @@
+package ru.nsu.fit.directors.userservice.event;
+
+public record BusinessMessageEvent(Long userId, Long orderId, String message) {
+}

src/main/java/ru/nsu/fit/directors/userservice/listener/ChatTopicListener.java

@@ -0,0 +1,23 @@
+package ru.nsu.fit.directors.userservice.listener;
+
+import javax.annotation.ParametersAreNonnullByDefault;
+
+import lombok.RequiredArgsConstructor;
+import org.springframework.kafka.annotation.KafkaHandler;
+import org.springframework.kafka.annotation.KafkaListener;
+import org.springframework.stereotype.Component;
+import ru.nsu.fit.directors.userservice.event.BusinessMessageEvent;
+import ru.nsu.fit.directors.userservice.service.ChatService;
+
+@Component
+@RequiredArgsConstructor
+@ParametersAreNonnullByDefault
+@KafkaListener(topics = "chatTopic")
+public class ChatTopicListener {
+    private final ChatService chatService;
+
+    @KafkaHandler
+    public void handleMessage(BusinessMessageEvent businessMessageEvent) {
+        chatService.handleMessage(businessMessageEvent);
+    }
+}

src/main/java/ru/nsu/fit/directors/userservice/service/ChatService.java

@@ -7,6 +7,7 @@
 
 import ru.nsu.fit.directors.userservice.dto.MessageDto;
 import ru.nsu.fit.directors.userservice.dto.request.ChatMessage;
+import ru.nsu.fit.directors.userservice.event.BusinessMessageEvent;
 
 @ParametersAreNonnullByDefault
 public interface ChatService {
@@ -24,4 +25,11 @@ public interface ChatService {
      */
     @Nonnull
     List<MessageDto> getChat(Long orderId);
+
+    /**
+     * Обработать входящее сообщение
+     *
+     * @param businessMessageEvent входящее сообщение
+     */
+    void handleMessage(BusinessMessageEvent businessMessageEvent);
 }

src/main/java/ru/nsu/fit/directors/userservice/service/ChatServiceImpl.java

@@ -8,10 +8,12 @@
 
 import lombok.RequiredArgsConstructor;
 import org.springframework.kafka.core.KafkaTemplate;
+import org.springframework.messaging.simp.SimpMessagingTemplate;
 import org.springframework.stereotype.Service;
 import ru.nsu.fit.directors.userservice.api.OrderServiceClient;
 import ru.nsu.fit.directors.userservice.dto.MessageDto;
 import ru.nsu.fit.directors.userservice.dto.request.ChatMessage;
+import ru.nsu.fit.directors.userservice.event.BusinessMessageEvent;
 import ru.nsu.fit.directors.userservice.event.UserMessageEvent;
 import ru.nsu.fit.directors.userservice.model.User;
 
@@ -23,6 +25,7 @@ public class ChatServiceImpl implements ChatService {
     private final KafkaTemplate<String, UserMessageEvent> kafkaTemplate;
     private final SecurityService securityService;
     private final OrderServiceClient orderServiceClient;
+    private final SimpMessagingTemplate simpMessagingTemplate;
 
     @Override
     public void save(ChatMessage chatMessage, Long orderId) {
@@ -39,4 +42,9 @@ public List<MessageDto> getChat(Long orderId) {
         User user = securityService.getLoggedInUser();
         return Objects.requireNonNull(orderServiceClient.getMessages(user.getId(), orderId).getBody()).getResult();
     }
+
+    @Override
+    public void handleMessage(BusinessMessageEvent businessMessageEvent) {
+        simpMessagingTemplate.convertAndSend("/topic/" + businessMessageEvent.orderId(), businessMessageEvent);
+    }
 }

src/main/resources/application.properties

@@ -34,7 +34,8 @@ spring.kafka.template.default-topic=orderTopic
 
 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.type.mapping=notification:ru.nsu.fit.directors.userservice.event.OrderNotificationEvent
+spring.kafka.consumer.properties.spring.json.type.mapping=notification:ru.nsu.fit.directors.userservice.event.OrderNotificationEvent,\
+  business-message:ru.nsu.fit.directors.userservice.event.BusinessMessageEvent
 
 springdoc.swagger-ui.path=/user/swagger-ui.html
 springdoc.api-docs.path=/user/api-docs