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 같은 리액티브 프로그래밍 라이브러리에서 지원한다고 보시면 될 것 같아요.