Downstream Mono, Flux가 subscriber인가요?

안녕하세요? Downstream Flux, Mono에 대해서 질문을 주셨는데요.아래 간단한 코드를 예로 들면(Mono도 같은 원리라고 보시면 됩니다.)Flux .just(1, 2, 4) .filter(num -> num % 2 == 0) .subscribe(result -> System.out.println(result));just()는 일반적으로 최초로 데이터를 emit하기 때문에 생산자, Publisher, 발행자로 이해하면 됩니다.그런데 filter() 입장에서는 just()가 Upstream Flux가 되고, 반대로 just() 입장에서는 filter()가 Downstream Flux가 됩니다.이렇게 부르는 이유는 just()와 filter() 같은 Operator들이 리턴하는 리턴 타입이 Mono 또는 Flux이기 때문입니다.즉, filter() 입장에서는 윗쪽에 있는 just()가 리턴하는 Flux를 이용해서 다음 흐름을 이어가기 때문에 just()가 Upstream Flux가 되는 것입니다. 그리고 Subscriber는 내부적으로 좀 복잡한 과정을 거치는데 다 이해하실 필요는 없고, 일반적으로 .subscribe(...) 여기에서 ... 에 해당되는 람다 표현식이 Subscriber라고 이해하셔도 무방할 것 같습니다. 혹시 제 설명이 좀 어려우시다면 just()와 filter() 같은 Operator 내부로 들어가셔서 리턴 타입을 직접 눈으로 확인하시면 좋을 것 같아요. 감사합니다!

onErrorResume을 사용하지 않는 모든 경우 예외 발생 시, 시퀀스는 종료되나요?

안녕하세요. 주말이라서 답변이 좀 늦었네요.onErrorResume 같은 에러 처리를 위한 Operator를 사용하지 않으면 기본적으로 Sequence 내에서 에러가 발생하면 Subscriber에게 onError Signal 형태로 에러 정보가 Exception 형태로 전달됩니다. 다만, onErrorResume을 사용하지는 않지만 특정 상황에서 onErrorContinue Operator를 사용하거나 retry Operator를 사용하면 에러 발생 시점에 Sequence가 즉시 중단되지 않을 수는 있습니다.방금 말씀 드린 내용은 error handling 섹션에 있는 영상으로 확인하실 수 있습니다. AI 인턴이 답변 남겨두었듯이 doOnError 같은 Operator를 이용하면 에러를 직접적으로 핸들링하는건 아니지만 에러가 발생했을 때, 로그를 남긴다든지 하는 후속 작업을 통해 에러를 간접적으로 알려서 추후에 디버깅을 할 수 있습니다. 디버깅 관련해서는 1부 영상을 확인하시면 될 것 같습니다.감사합니다.

source.next와 source.emit의 차이에 대한 질문입니다.

안녕하세요? 어제 일찍 잠이 드는 바람에 답변이 좀 늦어졌네요. 양해 부탁드리겠습니다. ^^; 우선 아래 예제를 실행시켜 보시면 next() 호출 후에 onComplete signal이 발생하지 않기 때문에 Sequence가 종료 되지 않고 무한 대기 상태가 되는걸 확인하실 수가 있는데요. TestPublisher source = TestPublisher.create(); StepVerifier .create(source.flux()) .expectSubscription() .then(() -> source.next(2, 4, 6, 8, 10)) .expectNext(2, 4, 6, 8, 10) .expectComplete() .verify(); 말씀하셨던대로 TestPublisherTestExample01에서 expectComplete()이 통과하는 이유는 zipWith()를 사용했기 때문입니다. 아래 그림을 보시면 zipWith()의 파라미터로 전달한 Flux에서 데이터 emit이 끝나면 onComplete signal이 발생하는데 zipWith()의 경우 둘 중 하나의 Sequenc가 종료될 경우 최종적으로 합쳐진 Source Flux에서 onComplete signal이 발생합니다.(사진)

StepVerifier를 이용한 Testing (1) - expectNoEvent 에 관해서

안녕하세요? expectNoEvent()의 의미가 조금 헷갈리시죠?테스트 하셨던 코드처럼 .expectNoEvent(Duration.ofMinutes(1)) 를 두 번 연달아 호출했다고 해서 2분의 딜레이 타임을 가진다라는 의미가 아니라 매 1분(여기서는 두 번 호출했으니까 총 2분이겠죠) 즉, 총 2분 동안 단지 어떠한 이벤트도 발생하지 않았는지를 단순히 검증만 하는 것이라서 .expectNoEvent(Duration.ofMinutes(1)) 를 호출했다고 해서 getVoteCount()에서 리턴한 Flux의 Tuple 데이터가 emit 되지는 않고 예제 코드에서처럼 expectNext()를 호출해야 데이터 하나가 emit이 됩니다.지금 테스트 해 보셨던 코드에서는 ... .expectNoEvent(Duration.ofMinutes(1)) .expectNoEvent(Duration.ofMinutes(1)) .expectNext(Tuples.of("강서구", 32040)) ... ....expectNoEvent(Duration.ofMinutes(1)) 를 두 번 호출 한 다음에 호출한 .expectNext()가 세번째 .expectNext() 이기 때문에 "강서구"가 맞으므로 테스트에 통과하는 것입니다. 그리고 아직 테스트 대상 메서드의 Flux에서 emit되지 않은 데이터가 남았는데 expectNext()나 expectNextCount() 등으로 남아있는 데이터의 emit에 해당하는 onNext 이벤트를 트리거 해주지 않으면 테스트 코드가 종료되지 않고 무한정 기다릴 수 있는 상황도 발생할 수 있으니 참고해 주시면 좋을 것 같아요.

backpressure latest 전략

안녕하세요.Discard Support: Each time a new element comes in (the new "latest"), this operator discards the previously retained element. 제가 알고있는 바로는 공식 문서에 나와 있는 이 문장의 의미는 매 시점마다 들어오는 데이터(new element) 중에서 바로 직전 시점에 남겨진 데이터를 discard하고 그 다음 시점에 들어온 데이터가 최신 데이터가 되어서 버퍼에 들어갈 공간이 있게되는 시점에 버퍼 안에 채워지는걸로 알고 있습니다.여기서 말하는 previously retained element는 들어오는 데이터들 중에서 가장 최근에 들어오는 데이터 중에서 오래된 데이터를 의미하는데, 버퍼를 다 비운다는 설명이 구체적으로 나와 있지는 않습니다. 말씀하신것과 조금 유사한 방식으로 버퍼 내부에서 어떤 데이터를 drop 할 것인지를 핸들링하는 전략인 onBackpressureBuffer()를 사용할 수 있습니다.

StepVerifier를 이용한 Testing (1) 의 StepVerifiter 질문

안녕하세요. 답변이 조금 늦어서 죄송합니다.전에 비슷한 질문에 답변 드린적이 있어서 해당 질문의 답변 내용 확인하시면 도움이 되실것 같아요.아래 링크 확인해주시면 감사드리겠습니다.https://www.inflearn.com/community/questions/1286668

[Scheduler의 종류 강의] Schedulers.newParallel 메서드에 관한 질문

안녕하세요.Spring WebFlux에서 work-stealing 같은 처리를 지원하는지 질문 주셨는데요.제가 알기로는 아래와 같이 ParallelFlux가 work-stealing 기능을 지원은 하는걸로 알고 있습니다.(사진) 다만, 작업을 가져가서 처리하는 work-stealing이라는 일반적인 의미와는 조금 다른 의미라고 생각해요. 리액티브 프로그래밍에서는 들어오는 데이터 즉, 데이터 스트림을 처리할 때 라운드 로빈 방식으로 데이터를 처리하는데 처리할 데이터를 미리 fetch(prefetch)해서 작업 효율성을 높일 수 있는데요. 이 prefetch하는 데이터의 양을 내부적으로 동적으로 조절해서 처리하는걸로 알고 있습니다.한마디로 다른 쓰레드보다 들어오는 데이터를 더 빨리 처리하는 쓰레드는 데이터를 조금 더 많이 prefetch 해서 처리한다고 생각하시면 될 것 같습니다. Spring WebFlux에서 이런 기능을 지원한다기 보다는 Reactor 같은 리액티브 프로그래밍 라이브러리에서 지원한다고 보시면 될 것 같아요.

Backpressure Drop 전략에서 다시 버퍼가 채워지는 시점에 대한 질문입니다.

아, 이건 저도 내부 깊숙히 들어가서 찾아 본건 아닌것 같습니다. 기억이 안나는걸 보면요. ㅡㅡㅋ다만 문서에서 확인했었습니다. 깊게 찾아 들어갈 필요성을 느끼질 못해서 안해봤었습니다. ㅡ.,ㅡ 오늘은 밤이 늦어서 나중에 시간될 때 저도 찾아보도록 하겠습니다.감사합니다.

Reactor 3부의 오픈 일정에 관해서 문의드립니다!

사실 3부 이후도 머리속에 기획은 대부분 되어 있는데, 제가 지금 오픈 준비 중인 강의 이후로 일정을 계획하고 있어서 시간이 조금 걸릴 것 같습니다.그래서 죄송스런 마음입니다. ^^;Non-Blocking I/O 방식의 애플리케이션에서 핵심은 사실 Reactor라서 Reactor를 우선 열심히 학습해 주시면 감사드릴게요.

inner sequence context 관련 질문

안녕하세요.Context 동작 방식이 좀 헷갈리시죠? ^^;이해하기 힘드실까봐 코드에 그림을 잠깐 그려봤습니다.(사진)(1)번 같은 경우에는 Inner Sequence에서 외부 시퀀스를 읽는 것이므로 정상적으로 itvillage라는 값을 읽어옵니다.(2)번 같은 경우에는 Inner Sequece 안에서 내부 시퀀스를 읽는 것이므로 역시 정상적으로 Software Engineer라는 값을 읽어오구요. 만약에 Inner Sequence 내부의 컨텍스트를 Inner Sequence 바깥 쪽에서 읽으면 값을 읽을 수 없습니다.Inner Sequence 내부에서 읽어서 다시 emit을 해 버리기 때문에 출력이 되는건데 값이 출력이 되어 버리니까 더 헷갈리셨던것 같네요.강의 영상에서 네 번째 예제 코드 뒷쪽 설명 부분에 가면 외부에서 Inner Sequence 내부에 있는 컨텍스트를 읽을 수 없는 부분에 대해서 설명 드리고 있으니 참고하시면 될 것 같아요. 답변이 되셨으면 좋겠습니다. 감사합니다.