Spring Boot, WebFlux 에서 Reactive Redis Sorted Set 연동

CREATE TABLE `member` (
  `id` int(11) unsigned NOT NULL AUTO_INCREMENT,
  `name` varchar(128) NOT NULL,
  `age` int(2) NOT NULL,
  `created_at` datetime NOT NULL,
  PRIMARY KEY (`id`)
) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_general_ci;

src
└── main
    └── java
        └── com.example.webfluxapi
         	├── common
         	├── controller
          	├── dto
        	├── entity
         	├── repository
         	└── service
		        └── impl

package com.example.webfluxapi.service;

import com.example.webfluxapi.dto.MemberDTO;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public interface MemberService {
    public Mono<MemberDTO.Crud> create(MemberDTO.Crud dto);
    public Mono<MemberDTO.Crud> item(Integer id);
    public Flux<MemberDTO.Crud> list(Integer page, Integer limit);
}

package com.example.webfluxapi.service.impl;

import com.example.webfluxapi.dto.MemberDTO;
import com.example.webfluxapi.mapper.MemberMapper;
import com.example.webfluxapi.repository.MemberRepository;
import com.example.webfluxapi.service.MemberService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;

@Slf4j
@Service
@ConditionalOnProperty(name = "redis.cache", havingValue = "false")
public class MemberServiceDB implements MemberService {

    private final MemberRepository memberRepository;
    private final MemberMapper memberMapper;

    @Autowired
    public MemberServiceDB(MemberRepository memberRepository, MemberMapper memberMapper) {
        this.memberRepository = memberRepository;
        this.memberMapper = memberMapper;
    }

    @Override
    public Mono<MemberDTO.Item> create(MemberDTO.Create dto) {
        dto.setCreatedAt(LocalDateTime.now());
        return memberRepository.save(memberMapper.ToCreateEntity(dto)).map(memberMapper::ToDTOItem);
    }

    @Override
    public Mono<MemberDTO.Item> item(Integer id) {
        return memberRepository.findById(id).map(memberMapper::ToDTOItem);
    }

    @Override
    public Flux<MemberDTO.Item> list(Integer page, Integer limit) {
        return memberRepository.findAllBy((page - 1) * limit, limit).map(memberMapper::ToDTOItem);
    }
}

package com.example.webfluxapi.service.impl;

import com.example.webfluxapi.dto.MemberDTO;
import com.example.webfluxapi.mapper.MemberMapper;
import com.example.webfluxapi.repository.MemberRepository;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.autoconfigure.condition.ConditionalOnProperty;
import org.springframework.data.domain.Range;
import org.springframework.data.redis.connection.Limit;
import org.springframework.data.redis.core.ReactiveRedisTemplate;
import org.springframework.data.redis.core.ReactiveZSetOperations;
import org.springframework.stereotype.Service;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

@Slf4j
@Service
@ConditionalOnProperty(name = "redis.cache", havingValue = "true")
public class MemberServiceCache extends MemberServiceDB {

    private final ObjectMapper objectMapper = new ObjectMapper();
    private final String KEY = "member";
    private final ReactiveZSetOperations<String, String> reactiveRedisZSet;

    @Autowired
    public MemberServiceCache(
            MemberRepository memberRepository,
            MemberMapper memberMapper,
            ReactiveRedisTemplate<String, String> reactiveRedisTemplate
    ) {
        super(memberRepository, memberMapper);

        this.objectMapper.registerModule(new com.fasterxml.jackson.datatype.jsr310.JavaTimeModule());

        this.reactiveRedisZSet = reactiveRedisTemplate.opsForZSet();
    }

    @Override
    public Mono<MemberDTO.Crud> create(MemberDTO.Crud dto) {
        return super.create(dto).flatMap(
                member -> {
                    String encode;
                    try {
                        encode = objectMapper.writeValueAsString(member);
                    } catch (JsonProcessingException e) {
                        return Mono.error(new RuntimeException(e));
                    }
                    return this.reactiveRedisZSet
                            .add(KEY, encode, member.getId())
                            .then(Mono.just(member));
                }
        );
    }

    @Override
    public Mono<MemberDTO.Crud> item(Integer id) {
        return Mono.from(this.reactiveRedisZSet
                .rangeByScore(KEY, Range.from(Range.Bound.inclusive((double) id)).to(Range.Bound.inclusive((double) id)))
                .handle((member, sink) -> {
                    try {
                        sink.next(objectMapper.readValue(member, MemberDTO.Crud.class));
                    } catch (JsonProcessingException e) {
                        sink.error(new RuntimeException(e));
                    }
                })
        );
    }

    @Override
    public Flux<MemberDTO.Crud> list(Integer page, Integer limit) {
        return this.reactiveRedisZSet
                .reverseRangeByScore(
                        KEY,
                        Range.unbounded(),
                        Limit.limit().offset((page-1)*limit).count(limit)
                ).handle((member, sink) -> {
                    try {
                        sink.next(objectMapper.readValue(member, MemberDTO.Crud.class));
                    } catch (JsonProcessingException e) {
                        sink.error(new RuntimeException(e));
                    }
                });

    }
}