@EnableEurekaClient
@SpringBootApplication
@EnableFeignClients
@EnableResourceServer           // 보호 자원으로 설정
@EnableBinding(Sink.class)      // 이 애플리케이션을 메시지 브로커와 바인딩하도록 스프링 클라우드 스트림 설정
                                // Sink.class 로 지정 시 해당 서비스가 Sink 클래스에 정의된 채널들을 이용해 메시지 브로커와 통신
public class EventServiceApplication {
    // ... 생략

    /**
     * 레디스 서버에 실제 DB 커넥션을 설정
     * 레디스 인스턴스와 통신하려면 JedisConnectionFactory 를 빈으로 노출해야 함
     * 이 커넥션을 사용해서 스프링 RedisTemplate 객체 생성
     */
    @Bean
    public JedisConnectionFactory jedisConnectionFactory() {
        JedisConnectionFactory jedisConnectionFactory = new JedisConnectionFactory();
        jedisConnectionFactory.setHostName(customConfig.getRedisServer());
        jedisConnectionFactory.setPort(customConfig.getRedisPort());
        return jedisConnectionFactory;
    }

    /**
     * 레디스 서버에 작업 수행 시 사용할 RedisTemplate 객체 생성
     */
    @Bean
    public RedisTemplate<String, Object> redisTemplate() {
        RedisTemplate<String, Object> redisTemplate = new RedisTemplate<>();
        redisTemplate.setConnectionFactory(jedisConnectionFactory());
        return redisTemplate;
    }
}

/**
 * 레디스에 액세스해야 하는 클래스에 주입된 인터페이스
 */
public interface MemberRedisRepository {
    void saveMember(Member member);
    void updateMember(Member member);
    void deleteMember(String userId);
    Member findMember(String userId);
}

/**
 * 부트스트랩 클래스에서 정의한 RedisTemplate 빈을 사용하여 레디스 서버와 통신
 */
@Repository
public class MemberRedisRepositoryImpl implements MemberRedisRepository {

    private static final String HASH_NAME = "member";       // 회원 데이터가 저장되는 레디스 서버의 해시명
    private final RedisTemplate<String, Member> redisTemplate;
    private HashOperations hashOperations;      // HashOperation 클래스는 레디스 서버에 데이터 작업을 수행하는 스프링 헬퍼 메서드의 집합

    public MemberRedisRepositoryImpl(RedisTemplate redisTemplate) {
        this.redisTemplate = redisTemplate;
    }

    @PostConstruct
    public void init() {
        hashOperations = redisTemplate.opsForHash();

        // 키와 값을 명시적으로 직렬화해주지 않으면 default serializer 로 JdkSerializationRedisSerializer 를 사용하는데
        // 그러면 \xac\xed\x00\x05t\x00\x06member 이런 식으로 저장됨
       redisTemplate.setKeySerializer(new StringRedisSerializer());
        redisTemplate.setValueSerializer(new GenericJackson2JsonRedisSerializer());
    }

    @Override
    public void saveMember(Member member) {
        hashOperations.put(HASH_NAME, member.getId(), member);
    }

    @Override
    public void updateMember(Member member) {
        hashOperations.put(HASH_NAME, member.getId(), member);
    }

    @Override
    public void deleteMember(String userId) {
        hashOperations.delete(HASH_NAME, userId);
    }

    @Override
    public Member findMember(String userId) {
        return (Member) hashOperations.get(HASH_NAME, userId);
    }
}

/**
 * 회원 데이터 필요 시 회원 서비스 호출 전 레디스 캐시 먼저 확인
 */
@Component
public class MemberCacheRestTemplateClient {

    private static final Logger logger = LoggerFactory.getLogger(MemberCacheRestTemplateClient.class);

    private final RestTemplate restTemplate;
    private final MemberRedisRepository memberRedisRepository;
    private final CustomConfig customConfig;

    public MemberCacheRestTemplateClient(RestTemplate restTemplate, MemberRedisRepository memberRedisRepository,  CustomConfig customConfig) {
        this.restTemplate = restTemplate;
        this.memberRedisRepository = memberRedisRepository;
        this.customConfig = customConfig;
    }

    String URL_PREFIX = "/api/mb/member/";      // 회원 서비스의 Zuul 라우팅경로와 회원 클래스 주소

    /**
     * 회원 아이디로 레디스에 저장된 Member 클래스 조회
     */
    private Member checkRedisCache(String userId) {
        try {
            return memberRedisRepository.findMember(userId);
        } catch (Exception e) {
            logger.error("======= Error encountered while trying to retrieve member {} check Redis Cache., Exception {}", userId, e);
            return null;
        }
    }

    /**
     * 레디스 캐시에 데이터 저장
     */
    private void cacheMemberObject(Member member) {
        try {
            memberRedisRepository.saveMember(member);
        } catch (Exception e) {
            e.printStackTrace();
            logger.error("======= Unable to cache member {} in Redis. Exception {}", member.getId(), e);
        }
    }

    public Member getMember(String userId) {

        Member member = checkRedisCache(userId);

        // 레디스에 데이터가 없다면 원본 데이터에서 데이터를 조회하기 위해 회원 서비스 호출
        if (member != null) {
            logger.debug("======= Successfully retrieved an Member {} from the redis cache: {}", userId, member);
            return member;
        }

        logger.debug("======= Unable to locate member from the redis cache: {}", userId);

        ResponseEntity<Member> restExchange =
                restTemplate.exchange(
                        "http://" + customConfig.getServiceIdZuul() + URL_PREFIX + "{userId}",   // http://localhost:5555/api/mb/member/userInfo/rinda
                        HttpMethod.GET,
                        null,
                        Member.class,
                        userId
                );

        // 캐시 레코드 저장
        member = restExchange.getBody();

        // 조회한 객체를 캐시에 저장
        if (member != null) {
            cacheMemberObject(member);
        }
        return member;
    }
}

// ... 생략
private final MemberCacheRestTemplateClient memberCacheRestTemplateClient;

/**
 * 레디스 캐싱 데이터 사용
 */
@GetMapping(value = "{userId}")
public Member userInfo(@PathVariable("userId") String userId) {
    return memberCacheRestTemplateClient.getMember(userId);
}

/**
 * 사용자 정의 input 채널 (SINK.INPUT 같은...), Consumer
 */
public interface CustomChannels {

    @Input("inboundMemberChanges")      // @Input 은 채널 이름을 정의하는 메서드 레벨 애너테이션
    SubscribableChannel members();      // @Input 애너테이션으로 노출된 채널은 모두 SubscribableChannel 클래스를 반환해야 함
}

/**
 * 사용자 정의 채널을 사용하여 메시지 수신
 * 이 애플리케이션을 메시지 브로커와 바인딩하도록 스프링 클라우드 스트림 설정
 */
@EnableBinding(CustomChannels.class)    // CustomChannels.class 로 지정 시 해당 서비스가 CustomChannels 클래스에 정의된 채널들을 이용해 메시지 브로커와 통신
public class MemberChangeHandler {

    private static final Logger logger = LoggerFactory.getLogger(MemberChangeHandler.class);
    private final MemberRedisRepository memberRedisRepository;

    public MemberChangeHandler(MemberRedisRepository memberRedisRepository) {
        this.memberRedisRepository = memberRedisRepository;
    }

    /**
     * 메시지가 입력 채널에서 수신될 때마다 이 메서드 실행
     */
    @StreamListener("inboundMemberChanges")     // Sink.INPUT 대신 사용자 정의 채널명인 inboundMemberChanges 전달
    public void loggerSink(MemberChangeModel mbChange) {
        logger.info("======= Received a message of type {}", mbChange.getType());
        switch (mbChange.getAction()) {
            case "GET":
                logger.debug("Received a GET event from the member service for userId {}", mbChange.getUserId());
                break;
            case "SAVE":
                logger.debug("Received a SAVE event from the member service for userId {}", mbChange.getUserId());
                break;
            case "UPDATE":
                logger.debug("Received a UPDATE event from the member service for userId {}", mbChange.getUserId());
                memberRedisRepository.deleteMember(mbChange.getUserId());       // 캐시 무효화
                break;
            case "DELETE":
                logger.debug("Received a DELETE event from the member service for userId {}", mbChange.getUserId());
                memberRedisRepository.deleteMember(mbChange.getUserId());
                break;
            default:
                logger.debug("Received an UNKNOWN event from the member service for userId {}", mbChange.getType());
                break;
        }
    }
}

// 부트스트랩에 작업했었던 채널 매핑
@StreamListener(Sink.INPUT)     // 메시지가 입력 채널에서 수신될 때마다 이 메서드 실행

// 사용자 정의 채널 매핑
@StreamListener("inboundMemberChanges")     // Sink.INPUT 대신 사용자 정의 채널명인 inboundMemberChanges 전달

/**
 * 단순 메시지 발행
 */
@PostMapping("/{userId}")
public void saveUserId(@PathVariable("userId") String userId) {
    // DB 에 save 작업..
    simpleSourceBean.publishMemberChange("SAVE", userId);
}

/**
 * 이벤트 서비스에서 캐시 제거를 위한 메서드
 */
@DeleteMapping("userInfo/{userId}")
public void deleteUserInfoCache(@PathVariable("userId") String userId) {
    logger.debug("====== 회원 삭제 후 DELETE 메시지 발생");

    // DB 에 삭제 작업  (간편성을 위해 DB 작업은 생략)
    simpleSourceBean.publishMemberChange("DELETE", userId);
}