Cold Sequence & Hot Sequence

✏️ Cold Sequence

📍 기본 개념

• Subscriber 가 구독할 때마다 타임라인이 생성 처음부터 emit 된 모든 데이터를 받을 수 있다.
  • 즉, 구독 시점이 달라도 모든 Subscriber 의 데이터가 같다.
  • 또한 각각의 Subscriber 는 각각의 타임라인을 갖기때문에 구독횟수만큼의 타임라인이 존재한다.

📍 적용

• 구독 시점에 상관없이 모두 같은 data 를 갖게됨

public static void main(String[] args) {
    Flux<String> coldFlux = Flux.fromIterable(Arrays.asList("RED", "YELLOW", "PINK"))
            .map(String::toLowerCase);

    coldFlux.subscribe(country -> log.info("구독 1 : {}", country));
    log.info("-------------------------------");
    coldFlux.subscribe(country -> log.info("구독 2 : {}", country));
}

구독 1 : red
구독 1 : yellow
구독 1 : pink
----------------------
구독 2 : red
구독 2 : yellow
구독 2 : pink

✏️ Hot Sequence

📍 기본 개념

• Subscriber 가 구독한 시점의 타임라인부터 emit 된 데이터를 받을 수 있다.
  • 타임라인이 구독횟수와 상관없이 하나만 존재한다.

📍 적용

• delayElements() data 를 순차적으로 publish 하는 메서드
• share() 타임라인을 공유해 Hot Sequence 로 변경하는 메서드

public static void main(String[] args) throws InterruptedException {
    Flux<String> concertflux = Flux
            .fromStream(Stream.of("Singer A", "Singer B", "Singer C", "Singer D", "Singer E"))
            // delay 를 사용해 1초에 한번씩 data 를 publish 하도록 설정
            .delayElements(Duration.ofSeconds(1))
            // 하나의 타임라인으로 다수의 subscriber 가 공유하도록 설정
            .share();

    concertflux.subscribe(singer -> System.out.println("구독 1 : " + singer));

    Thread.sleep(2500);

    concertflux.subscribe(singer -> System.out.println("구독 2 : " + singer));

    // 요청이 종료되지 않고 모든 타임라인이 끝날 때 까지 지연시킴
    Thread.sleep(3000);
}

구독 1 : Singer A
구독 1 : Singer B
구독 1 : Singer C
구독 2 : Singer C
구독 1 : Singer D
구독 2 : Singer D
구독 1 : Singer E
구독 2 : Singer E