Cold Sequence와 Hot Sequence

Cold Sequence

• Seuqnece는 Publisher가 emit하는 데이터의 연속적인 흐름을 정의해 놓은 것으로 코드로 표현하면 Operator 체인 형태로 정의됩니다.
• Cold Sequence는 Subscribe가 구독할 때마다 데이터 흐름이 처음부터 다시 시작되는 Sequence입니다.
• 즉, Subsriber의 구독 시점이 달라도 구독을 할 때마다 Publisher가 데이터를 emit하는 과정을 처음부터 다시 시작하는 데이터의 흐름을 Cold Sequence라고 부릅니다.

public static void main(String[] args) throws InterruptedException {
    Flux<String> coldFlux =
            Flux.fromIterable(Arrays.asList("KOREA", "JAPAN", "CHINESE")) ;

    coldFlux.subscribe(System.out::println) ;

    Thread.sleep(2000);
    System.out.println();
    coldFlux.subscribe(System.out::println) ;
}

• 위 코드를 실행하면 KOREA, JAPAN, CHINESE가 출력된 다음, 다시 KOREA, JAPAN, CHINESE가 출력된다.

---

Hot Sequence

• Hot Sequence의 경우 구독이 발생한 시점 이전에 Pub로부터 emit된 데이터는 Sub가 전달바딪 못하고 구독이 발생한 시점 이후에 emit된 데이터만 전달받을 수 있다.

public static void main(String args[]) {
	String[] singer = {
    	"Singer A",
        "Singer B",
        "Singer C",
        "Singer D",
        "Singer E"
    } ;
    
    log.info("# Begin concert:") ;
    
    Flux<String> concertFlux =
    	Flux.fromArray(singers)
            .delayElements(Duration.ofSeconds(1))
            .share() ;
        
        concertFlux.subscribe(
        	singer -> log.info("# Subscriber1 is watching {}'s song", singer)
        ) ;
        
        Thread.sleep(2500) ;
        
        concertFlux.subscribe(
        	singer -> log.info("# Subscriber2 is watching {}'s song", singer)
        ) ;
        
        Thread.sleep(3000) ;
}

 

• delayElements() Operator는 데이터 소스로 입력된 각 데이터의 emit을 일정시간 동안 지연시키는 Operator입니다.
• share() Operator는 Cold Sequence를 Hot Sequence로 동작하게 해 주는 Operator입니다.