@Builder
public record Chat(
        String user,
        String message
) {
}

@Configuration
@EnableWebSocket
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketConfigurer {

    private final WebSocketHandler webSocketHandler; // 커스텀 클래스

    @Override
    public void registerWebSocketHandlers(WebSocketHandlerRegistry registry) {
        registry.addHandler(webSocketHandler, "/websocket")
                .setAllowedOrigins("*");
    }
}

@Slf4j
@Component
public class WebSocketHandler extends TextWebSocketHandler {

    // 세션 List(Set)
    private static Set<WebSocketSession> sessions = new HashSet<WebSocketSession>(); 
    private final ObjectMapper mapper = new ObjectMapper();

    /* 클라이언트(세션) 연결 시 */
    @Override
    public void afterConnectionEstablished(WebSocketSession session) throws Exception {
        sessions.add(session);
    }

    /* 클라이언트(세션)가 메시지를 송신할 경우 */
    @Override
    public void handleMessage(WebSocketSession session, WebSocketMessage<?> message) throws IOException {

        log.info("payload : {}", message.getPayload());
        Chat chat = mapper.readValue(message.getPayload().toString(), Chat.class);
        log.info("chat : {}", chat);
        log.info("user : {}", chat.user());
        log.info("message : {}", chat.message());


        // 세션 List(Set)에 등록된 모든 세션에들에게 메시지 전달
        sessions.forEach(s -> {
            try{
                s.sendMessage(new TextMessage(chat.user() + ": " + chat.message()));
            }
            catch(IOException e){
                log.error("error : {}", e.getMessage());
            }
        });
    }

    /* 클라이언트(세션) 연결 종료 시 */
    @Override
    public void afterConnectionClosed(WebSocketSession session, CloseStatus status) throws Exception {
        sessions.remove(session);
    }
}

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {

    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/sub"); // 특정 토픽 구독 엔드포인트 접두사 설정
        registry.setApplicationDestinationPrefixes("/pub"); // 메시지 발행 시 엔드포인트 접두사 설정
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/stomp")
                .setAllowedOriginPatterns("*")
                .withSockJS(); // SockJs로 연결 시 http://localhost:8080/stomp로 연결
    }
}

@Controller
@RequiredArgsConstructor
@Slf4j
public class MessageController {

    @MessageMapping("{stockId}")    // 앞에 pub 접두사가 자동으로 붙음, 메시지를 발행 pub
    @SendTo("/sub/stocks")          // 특정 토픽을 sub한 클라이언트들에게 전달
    public StockInfo message(@DestinationVariable String stockId) {
        log.info("stock id: {}", stockId);
        return StockInfo.builder()
                .stockId(stockId)
                .price(2000)
                .build();
    }
}

// StockInfo.java
@Builder
public record StockInfo(
        String stockId,
        int price
) {
}

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/sub"); // 특정 토픽을 구독하는 경로
        
        /* 클라이언트가 메시지를 발행(pub)하지 않으니 발행 경로 추가 X */
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/stomp")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
}

@Configuration
@EnableScheduling // 스케줄링 기능 사용을 위하여 추가
public class SchedulingConfig {
}

@Component
@RequiredArgsConstructor
public class MessageSender {

    private final SimpMessagingTemplate messagingTemplate;

    public void sendMessage() {
        messagingTemplate.convertAndSend("/sub/stocks", getStockInfo("apple", 2000));
    }

    public StockInfo getStockInfo(String stockId, int price) {
        return StockInfo.builder()
                .stockId(stockId)
                .price(price)
                .build();
    }
}

@Component
@RequiredArgsConstructor
@Slf4j
public class MessageScheduler {

    private final MessageSender messageSender;

    @Scheduled(fixedRate = 5000)
    public void scheduleMessage(){
        log.info("Message scheduled");
        messageSender.sendMessage();
    }
}

implementation 'org.springframework.boot:spring-boot-starter-data-redis'

@Configuration
@RequiredArgsConstructor
public class RedisConfig {

    @Value("${spring.data.redis.host}")
    private String host;
    @Value("${spring.data.redis.port}")
    private int port;

    @Bean
    public RedisConnectionFactory redisConnectionFactory() {
        RedisStandaloneConfiguration redisStandaloneConfiguration = new RedisStandaloneConfiguration(host, port);

        return new LettuceConnectionFactory(redisStandaloneConfiguration);
    }

    @Bean
    public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {

        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(redisConnectionFactory);
        redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
        redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer(ChatInfo.class)); // ❗ 이후 설명
        redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer(StockInfo.class)); // ❗ 이후 설명
        redisTemplate.setHashKeySerializer(new StringRedisSerializer());
        redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

        return redisTemplate;
    }

	/* MessageListener와 각 Topic들을 연결하는 설정을 가지는 Container Bean 등록 */
    @Bean
    public RedisMessageListenerContainer redisMessageListenerContainer(
            MessageListenerAdapter stockListener,
            MessageListenerAdapter chatListener,
            ChannelTopic stockTopic,
            ChannelTopic chatTopic
    ) {
        RedisMessageListenerContainer container = new RedisMessageListenerContainer();
        container.setConnectionFactory(redisConnectionFactory());
        
        // 주식 MessageListener와 주식 Topic을 연결
        container.addMessageListener(stockListener, stockTopic); 
        
        // 채팅 MessageListener와 주식 Topic을 연결
        container.addMessageListener(chatListener, chatTopic);
        
        return container;
    }

    /* 주식 메시지 브로커에서 보낸 메시지를 수신하기 위한 Adapter Bean 등록 */
    @Bean(name = "stockListener")
    public MessageListenerAdapter listenerAdapter(StockSubscriber subscriber) {
        return new MessageListenerAdapter(subscriber, "onMessage");
        // 두 번째 인자 = 레디스에서 보낸 메시지를 받아 처리할 메서드 이름 = "onMessage" 
    }

	/* 채팅 메시지 브로커에서 보낸 메시지를 수신하기 위한 Adapter Bean 등록 */
    @Bean(name = "chatListener")
    public MessageListenerAdapter chatListenrAdapter(ChatSubscriber subscriber) {
        return new MessageListenerAdapter(subscriber, "onMessage"); 
        // 두 번째 인자 = 레디스에서 보낸 메시지를 받아 처리할 메서드 이름 = "onMessage" 
    }

    /* 채널 주제 등록 - 주식 */
    @Bean(name = "stockTopic")
    public ChannelTopic stockTopic() {
        return new ChannelTopic("stock");
    }

	/* 채널 주제 등록 - 채팅 */
    @Bean(name = "chatTopic")
    public ChannelTopic chatTopic() {
        return new ChannelTopic("chat");
    }
}

@Configuration
@EnableWebSocketMessageBroker
@RequiredArgsConstructor
public class WebSocketConfig implements WebSocketMessageBrokerConfigurer {
    @Override
    public void configureMessageBroker(MessageBrokerRegistry registry) {
        registry.enableSimpleBroker("/sub"); // 특정 토픽을 구독
        registry.setApplicationDestinationPrefixes("/pub");  // 메시지를 발행
    }

    @Override
    public void registerStompEndpoints(StompEndpointRegistry registry) {
        registry.addEndpoint("/stomp")
                .setAllowedOriginPatterns("*")
                .withSockJS();
    }
}

@Builder
public record ChatInfo(
        String user,
        String message
){
}

@RestController
@RequiredArgsConstructor
@Slf4j
public class ChatPublisher {

    private final RedisTemplate redisTemplate;

    @MessageMapping("chat")
    public void pubChat(ChatInfo chatInfo) {
        log.info("pub message: {}", chatInfo.toString());

        redisTemplate.convertAndSend("chat", chatInfo);
    }
}

@Component
@Slf4j
@RequiredArgsConstructor
public class ChatSubscriber implements MessageListener {

    private final ObjectMapper mapper = new ObjectMapper();
    private final RedisTemplate redisTemplate;
    private final SimpMessageSendingOperations simpMessageSendingOperations;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        try {
        
          String listenedMessage = (String) redisTemplate.getStringSerializer().deserialize(message.getBody());

          log.info("listened message: {}", listenedMessage);
          ChatInfo chatInfo = mapper.readValue(listenedMessage, ChatInfo.class);

          simpMessageSendingOperations.convertAndSend("/sub/chat", chatInfo);
        
        } catch (JsonProcessingException e) {
            log.error("error: {}", e.getMessage());
        }

    }
}

// RedisConfig.java
@Bean
public RedisTemplate<String, Object> redisTemplate(RedisConnectionFactory redisConnectionFactory) {
    RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
    redisTemplate.setConnectionFactory(redisConnectionFactory);
    redisTemplate.setKeySerializer(new StringRedisSerializer());
    redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
    
    // ChatInfo.class에 관한 Serializer 추가
    redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer(ChatInfo.class));
    redisTemplate.setValueSerializer(new Jackson2JsonRedisSerializer(StockInfo.class));
    
    redisTemplate.setHashKeySerializer(new StringRedisSerializer());
    redisTemplate.setHashValueSerializer(new GenericJackson2JsonRedisSerializer());

    return redisTemplate;
}

@RequiredArgsConstructor
@Slf4j
@Component
public class StockPublisher {
    private final RedisTemplate redisTemplate;

    @Scheduled(fixedRate = 5000)
    public void publishStock() {
        log.info("pub stock");
        redisTemplate.convertAndSend("stock", getStockInfo());
    }

    private StockInfo getStockInfo(){
        return StockInfo.builder()
                .stockId("apple")
                .price(2000)
                .build();
    }
}

@Component
@Slf4j
@RequiredArgsConstructor
public class StockSubscriber implements MessageListener {

    private final ObjectMapper mapper = new ObjectMapper();
    private final RedisTemplate redisTemplate;
    private final SimpMessageSendingOperations simpMessageSendingOperations;

    @Override
    public void onMessage(Message message, byte[] pattern) {
        String listenedMessage = (String) redisTemplate.getStringSerializer().deserialize(message.getBody());

        log.info("listened message: {}", listenedMessage);

        try {
            StockInfo stockInfo = mapper.readValue(listenedMessage, StockInfo.class);

            simpMessageSendingOperations.convertAndSend("/sub/stocks", stockInfo);
        } catch (JsonProcessingException e) {
            log.error("error: {}", e.getMessage());
        }
    }
}