Merge pull request #9 from lotteon2/develop

Develop

build.gradle

@@ -31,14 +31,12 @@ dependencies {
     implementation 'org.springframework.cloud:spring-cloud-starter-zipkin:2.2.3.RELEASE'
     implementation 'org.springframework.cloud:spring-cloud-starter-sleuth'
     implementation group: 'io.micrometer', name: 'micrometer-registry-prometheus'
-
+    implementation 'org.springframework.boot:spring-boot-starter-data-redis-reactive'
     implementation 'io.github.boostchicken:spring-data-dynamodb:5.2.3'
     implementation 'software.amazon.awssdk:dynamodb:2.1.0'
 
-    implementation 'org.springframework.boot:spring-boot-starter-rsocket'
     implementation 'org.springframework.boot:spring-boot-starter-validation'
 
-    implementation 'org.springframework.boot:spring-boot-starter-web'
     implementation 'org.springframework.boot:spring-boot-starter-webflux'
 
     implementation 'org.springframework.cloud:spring-cloud-starter-circuitbreaker-reactor-resilience4j'
@@ -47,7 +45,6 @@ dependencies {
     implementation 'org.springframework.cloud:spring-cloud-starter-openfeign'
     implementation 'org.springframework.cloud:spring-cloud-starter-bus-amqp'
     implementation platform('org.testcontainers:testcontainers-bom:1.19.3') //import bom
-
     compileOnly 'org.projectlombok:lombok'
 
     developmentOnly 'org.springframework.boot:spring-boot-devtools'

src/main/java/com/dailyon/auctionservice/AuctionServiceApplication.java

@@ -3,46 +3,66 @@
 import com.amazonaws.services.dynamodbv2.AmazonDynamoDBAsync;
 import com.amazonaws.services.dynamodbv2.datamodeling.DynamoDBMapper;
 import com.amazonaws.services.dynamodbv2.model.CreateTableRequest;
+import com.amazonaws.services.dynamodbv2.model.Projection;
 import com.amazonaws.services.dynamodbv2.model.ProvisionedThroughput;
 import com.amazonaws.services.dynamodbv2.util.TableUtils;
 import com.dailyon.auctionservice.document.Auction;
+import com.dailyon.auctionservice.document.BidHistory;
 import org.springframework.beans.factory.annotation.Autowired;
 import org.springframework.boot.SpringApplication;
 import org.springframework.boot.autoconfigure.SpringBootApplication;
 import org.springframework.cloud.client.discovery.EnableDiscoveryClient;
 import org.springframework.cloud.openfeign.EnableFeignClients;
+import org.springframework.context.annotation.Profile;
 
 import javax.annotation.PostConstruct;
+import javax.annotation.PreDestroy;
 import java.util.TimeZone;
 
 @EnableFeignClients
 @EnableDiscoveryClient
 @SpringBootApplication
 public class AuctionServiceApplication {
-    @Autowired
-    AmazonDynamoDBAsync dynamoDB;
+  @Autowired AmazonDynamoDBAsync dynamoDB;
 
-    @Autowired
-    DynamoDBMapper dynamoDBMapper;
+  @Autowired DynamoDBMapper dynamoDBMapper;
 
-    public static void main(String[] args) {
-        SpringApplication.run(AuctionServiceApplication.class, args);
-    }
+  public static void main(String[] args) {
+    SpringApplication.run(AuctionServiceApplication.class, args);
+  }
 
-    @PostConstruct
-    public void setTimezoneToSeoul() {
-        TimeZone.setDefault(TimeZone.getTimeZone("Asia/Seoul"));
-    }
+  @PostConstruct
+  public void setTimezoneToSeoul() {
+    TimeZone.setDefault(TimeZone.getTimeZone("Asia/Seoul"));
+  }
 
-    // TODO : document FIX 후 삭제
-    @PostConstruct
-    public void setDynamoDB() {
-        TableUtils.deleteTableIfExists(dynamoDB, dynamoDBMapper.generateDeleteTableRequest(Auction.class));
+  // TODO : document FIX 후 삭제
+  @PostConstruct
+  @Profile({"!test"})
+  public void setDynamoDB() {
+    TableUtils.deleteTableIfExists(
+        dynamoDB, dynamoDBMapper.generateDeleteTableRequest(Auction.class));
 
-        CreateTableRequest createTableRequest = dynamoDBMapper
-                .generateCreateTableRequest(Auction.class)
-                .withProvisionedThroughput(new ProvisionedThroughput(1L, 1L));
+    TableUtils.deleteTableIfExists(
+        dynamoDB, dynamoDBMapper.generateDeleteTableRequest(BidHistory.class));
 
-        TableUtils.createTableIfNotExists(dynamoDB, createTableRequest);
-    }
+    CreateTableRequest createTableRequest =
+        dynamoDBMapper
+            .generateCreateTableRequest(Auction.class)
+            .withProvisionedThroughput(new ProvisionedThroughput(10L, 10L));
+
+    CreateTableRequest createTableRequest2 =
+        dynamoDBMapper
+            .generateCreateTableRequest(BidHistory.class)
+            .withProvisionedThroughput(new ProvisionedThroughput(1L, 1L));
+
+    createTableRequest2
+        .getGlobalSecondaryIndexes()
+        .forEach(
+            idx ->
+                idx.withProvisionedThroughput(new ProvisionedThroughput(1L, 1L))
+                    .withProjection(new Projection().withProjectionType("ALL")));
+    TableUtils.createTableIfNotExists(dynamoDB, createTableRequest);
+    TableUtils.createTableIfNotExists(dynamoDB, createTableRequest2);
+  }
 }

src/main/java/com/dailyon/auctionservice/chat/WebSocketSessionsManager.java

@@ -0,0 +1,35 @@
+package com.dailyon.auctionservice.chat;
+
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.socket.WebSocketSession;
+import reactor.core.publisher.Flux;
+import reactor.core.publisher.Mono;
+import reactor.core.publisher.Sinks;
+
+import java.util.Map;
+import java.util.concurrent.ConcurrentHashMap;
+
+@Component
+public class WebSocketSessionsManager {
+  private final Map<String, WebSocketSession> sessions = new ConcurrentHashMap<>();
+  private final Sinks.Many<WebSocketSession> sessionsSink =
+      Sinks.many().multicast().onBackpressureBuffer();
+
+  public void addSession(String userId, WebSocketSession session) {
+    // 세션을 저장소에 추가
+    sessions.put(userId, session);
+    // Sink에 세션 추가
+    sessionsSink.tryEmitNext(session);
+  }
+
+  public void removeSession(String userId) {
+    // 세션을 저장소에서 제거
+    sessions.remove(userId);
+  }
+
+  public Mono<Void> broadcast(String message) {
+    return Flux.fromIterable(sessions.values())
+            .flatMap(session -> session.send(Mono.just(session.textMessage(message))))
+            .then(); // 모든 세션에 대한 send 작업이 완료될 때까지 기다림
+  }
+}

src/main/java/com/dailyon/auctionservice/chat/messaging/RedisChatMessageListener.java

@@ -0,0 +1,43 @@
+package com.dailyon.auctionservice.chat.messaging;
+
+import com.dailyon.auctionservice.controller.ChatHandler;
+import com.dailyon.auctionservice.dto.request.Message;
+import com.dailyon.auctionservice.util.ObjectStringConverter;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.redis.connection.ReactiveSubscription;
+import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
+import org.springframework.data.redis.listener.PatternTopic;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
+
+import static com.dailyon.auctionservice.config.ChatConstants.MESSAGE_TOPIC;
+
+@Slf4j
+@Component
+@Profile({"!test"})
+public class RedisChatMessageListener {
+
+  private final ReactiveStringRedisTemplate reactiveStringRedisTemplate;
+  private final ChatHandler chatWebSocketHandler;
+  private final ObjectStringConverter objectStringConverter;
+
+  public RedisChatMessageListener(
+      ReactiveStringRedisTemplate reactiveStringRedisTemplate,
+      ChatHandler chatWebSocketHandler,
+      ObjectStringConverter objectStringConverter) {
+    this.reactiveStringRedisTemplate = reactiveStringRedisTemplate;
+    this.chatWebSocketHandler = chatWebSocketHandler;
+    this.objectStringConverter = objectStringConverter;
+  }
+
+  public Mono<Void> subscribeMessageChannelAndPublishOnWebSocket() {
+    return reactiveStringRedisTemplate
+        .listenTo(new PatternTopic(MESSAGE_TOPIC))
+        .map(ReactiveSubscription.Message::getMessage)
+        .flatMap(message -> objectStringConverter.stringToObject(message, Message.class))
+        .filter(chatMessage -> !chatMessage.getMessage().isEmpty())
+        .flatMap(chatWebSocketHandler::sendMessage)
+        .then();
+  }
+}

src/main/java/com/dailyon/auctionservice/chat/messaging/RedisChatMessagePublisher.java

@@ -0,0 +1,69 @@
+package com.dailyon.auctionservice.chat.messaging;
+
+import com.dailyon.auctionservice.dto.request.Message;
+import com.fasterxml.jackson.core.JsonProcessingException;
+import com.fasterxml.jackson.databind.ObjectMapper;
+import lombok.RequiredArgsConstructor;
+import lombok.extern.slf4j.Slf4j;
+import org.springframework.context.annotation.Profile;
+import org.springframework.data.redis.core.ReactiveStringRedisTemplate;
+import org.springframework.data.redis.support.atomic.RedisAtomicInteger;
+import org.springframework.data.redis.support.atomic.RedisAtomicLong;
+import org.springframework.stereotype.Component;
+import reactor.core.publisher.Mono;
+
+import java.net.InetAddress;
+import java.net.UnknownHostException;
+
+import static com.dailyon.auctionservice.config.ChatConstants.MESSAGE_TOPIC;
+
+@Slf4j
+@Component
+@Profile({"!test"})
+@RequiredArgsConstructor
+public class RedisChatMessagePublisher {
+
+  private final ReactiveStringRedisTemplate reactiveStringRedisTemplate;
+  private final RedisAtomicInteger chatMessageCounter;
+  private final RedisAtomicLong activeUserCounter;
+  private final ObjectMapper objectMapper;
+
+  public Mono<Long> publishChatMessage(String message) {
+    Integer totalChatMessage = chatMessageCounter.incrementAndGet();
+    return Mono.fromCallable(
+            () -> {
+              try {
+                return InetAddress.getLocalHost().getHostName();
+              } catch (UnknownHostException e) {
+                log.error("Error getting hostname.", e);
+              }
+              log.info("inetAddress.getLocalHost().getHostName() {} ", InetAddress.getLocalHost().getHostName());
+              return "localhost";
+            })
+        .map(
+            hostName -> {
+              log.info("message -> {}", message);
+              String chatString = "EMPTY_MESSAGE";
+              try {
+                Message chatMessage = objectMapper.readValue(message, Message.class);
+                chatMessage.setActiveUserCount(activeUserCounter.get());
+                chatString = objectMapper.writeValueAsString(chatMessage);
+              } catch (JsonProcessingException e) {
+                log.error("Error converting ChatMessage {} into string", message, e);
+                log.error("Error converting ChatMessage {} into string", "chatMessage", e);
+              }
+              return chatString;
+            })
+        .flatMap(
+            chatString -> {
+              // Publish Message to Redis Channels
+              return reactiveStringRedisTemplate
+                  .convertAndSend(MESSAGE_TOPIC, chatString)
+                  .doOnSuccess(
+                      aLong ->
+                          log.debug(
+                              "Total of {} Messages published to Redis Topic.", totalChatMessage))
+                  .doOnError(throwable -> log.error("Error publishing message.", throwable));
+            });
+  }
+}

src/main/java/com/dailyon/auctionservice/common/webclient/client/ProductClient.java

@@ -0,0 +1,45 @@
+package com.dailyon.auctionservice.common.webclient.client;
+
+
+import com.dailyon.auctionservice.common.webclient.response.CreateProductResponse;
+import com.dailyon.auctionservice.dto.request.CreateAuctionRequest;
+import com.dailyon.auctionservice.dto.response.ReadAuctionDetailResponse;
+import lombok.RequiredArgsConstructor;
+import org.springframework.stereotype.Component;
+import org.springframework.web.reactive.function.client.WebClient;
+import reactor.core.publisher.Mono;
+
+import java.util.List;
+import java.util.stream.Collectors;
+
+@Component
+@RequiredArgsConstructor
+public class ProductClient {
+    private final WebClient webClient;
+    public Mono<ReadAuctionDetailResponse.ReadProductDetailResponse> readProductDetail(Long productId) {
+        return webClient.get()
+                .uri(uriBuilder -> uriBuilder.path("/products/id/{productId}").build(productId))
+                .retrieve()
+                .bodyToMono(ReadAuctionDetailResponse.ReadProductDetailResponse.class);
+    }
+
+    public void deleteProducts(String memberId, String role, List<Long> ids) {
+        String joinedIds = ids.stream().map(String::valueOf).collect(Collectors.joining(","));
+        webClient.delete()
+                .uri(uriBuilder -> uriBuilder.path("/admin/products?ids={joinedIds}").build(joinedIds))
+                .header("memberId", memberId)
+                .header("role", role)
+                .retrieve();
+    }
+
+    public Mono<CreateProductResponse> createProduct(String memberId, String role,
+                                                     CreateAuctionRequest.CreateProductRequest request) {
+        return webClient.post()
+                .uri(uriBuilder -> uriBuilder.path("/admin/products").build())
+                .header("memberId", memberId)
+                .header("role", role)
+                .body(Mono.just(request), CreateAuctionRequest.CreateProductRequest.class)
+                .retrieve()
+                .bodyToMono(CreateProductResponse.class);
+    }
+}

src/main/java/com/dailyon/auctionservice/common/feign/response/CreateProductResponse.java → src/main/java/com/dailyon/auctionservice/common/webclient/response/CreateProductResponse.java

@@ -1,4 +1,4 @@
-package com.dailyon.auctionservice.common.feign.response;
+package com.dailyon.auctionservice.common.webclient.response;
 
 import lombok.AllArgsConstructor;
 import lombok.Builder;

src/main/java/com/dailyon/auctionservice/config/ChatConstants.java

@@ -0,0 +1,8 @@
+package com.dailyon.auctionservice.config;
+
+public class ChatConstants {
+    public static final String MESSAGE_TOPIC = "MESSAGE";
+    public static final String MESSAGE_COUNTER_KEY = "TOTAL_MESSAGE_COUNT";
+    public static final String ACTIVE_USER_KEY = "ACTIVE_USER_COUNT";
+    public static final String WEBSOCKET_MESSAGE_MAPPING = "ws/chat/**";
+}