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 내부에 있는 컨텍스트를 읽을 수 없는 부분에 대해서 설명 드리고 있으니 참고하시면 될 것 같아요. 답변이 되셨으면 좋겠습니다. 감사합니다.

boundedElastic 관련 질문

안녕하세요. boundedElastic()을 사용했을 때의 Thread에 대해서 질문 주셨는데요.제가 아래 첨부한 공식 문서 설명에서 이해한 바로는 (3)과 같이 Thread Pool을 사용하는건 맞는데, Pool에 미리 만들어두고 사용된다기 보다는 일단 Thread 사용 시점이 되면 그 시점에 Thread를 생성해서 제공한 후에 다 쓰면 Pool에 반납을 합니다. 그리고 나서 재사용을 할테구요.그리고 일반적으로 동시에 실행될 수 있는 Thread 개수는 CPU 코어의 10배라고 (2)에 명시되어 있습니다. 참고로 Reactor 3.6.0 부터는 Virtual Thread도 사용할 수 있다고 (1)에 명시되어 있네요.질문 주신거에 답변이 되셨으면 좋겠습니다.감사합니다.(사진)

강의자료 관련 질문드립니다.

인프런 쪽에서 발생한 오류였고, 해결되었다는 답변을 방금 전달 받아서 확인해보니 해결이 되었네요. ^^강의 자료 잘 다운로드 받으시고, 열심히 학습하시길 바래보겠습니다~감사합니다.