Flux onComplete 중복 발생

[ghkdqhrbals]

Description

저는 아래의 bm-controller restAPI 를 통해 bm-agent 에 SSE 요청을 진행합니다. 그런데 onComplete 이벤트가 두 번 실행되는 문제가 발생했습니다.

일단 추측으로는 SSE 가 종료될 떄 한 번, Flux 가 종료될 때 한 번 실행되서 총 두 번의 중복 onComplete 이벤트가 발생한다고 생각합니다.

    @PostMapping("/api/groups/{group_id}/templates/{template_id}")
    @PreAuthorize("hasRole('USER')")
    public ResponseEntity send(@PathVariable("group_id") String groupId,
        @PathVariable("template_id") String templateId,
        @RequestParam(value = "action") String action) {
        log.info("Send action: {}", action);
        String userId = userContext.getCurrentUser().getId();

        ParameterizedTypeReference<ServerSentEvent<TestResult>> typeReference =
            new ParameterizedTypeReference<ServerSentEvent<TestResult>>() {
            };

        WebClient webClient = WebClient.create("http://localhost:8081");

        Flux<ServerSentEvent<TestResult>> eventStream = webClient.post()
            .uri("/api/templates/{template_id}?action={action}", templateId, action)
            .retrieve()
            .bodyToFlux(typeReference)
            .log(); // TODO need to remove

        eventStream.subscribe(event -> {
                TestResult testResult = event.data();
                messagingTemplate.convertAndSend("/topic/" + groupId + "/" + templateId, testResult);
            },
            error -> {
                log.error("Error receiving SSE: {}", error.getMessage());
            },
            () -> {
            // success
                log.info("Test completed!");
                messagingTemplate.convertAndSend("/topic/" + userId, "test completed!");
            });

        return ResponseEntity.ok().build();
    }

그래서 저는 Boolean isListening 변수를 메소드 내부에 선언하고 해당 변수가 true 면 로직 실행. false 면 noOp() 하도록 설정하고 싶습니다. mutex lock 이 걸린 Boolean 을 선언하고 이를 관리해야하죠...

[LeeJeongGi]

우선 오전에 두번찍히는 이유를 좀 살펴봤는데요, 한쓰레드에서 두번 호출되는건 아닌거 같아요! ctor-http-nio-1 // ctor-http-nio-8 각각 다른 쓰레드라서 Boolean 처리를 못한거 같습니다. 제가 엄청 빨리 시작 중지를 눌러도 쓰레드 자체는 두개가 생성이 되는데 이부분이 뭔가 잘못된게 아닐까,,,라는 추측.. ㅎㅎ

[LeeJeongGi]

그래서 문득 생각이 든건 http://localhost:8080/api/groups/userGroup/templates/67?action=start 처리에 대한 쓰레드 하나와 http://localhost:8080/api/groups/userGroup/templates/67?action=stop 처리에 대한 쓰레드 하나가 각각 종료되면서 "Test completed!" 가 두번 찍히는게 아닐까 생각합니다! :)

[ghkdqhrbals]

Wow!!! @LeeJeongGi 님이 말씀하신 부분을 실제 손버깅 해보니 맞았습니다 :) 감사합니다!

아래처럼 action 이 stop 일 떄와 start 일 떄 각각 API 호출을 진행하기 때문에 두 번 발생하는 것이였네요ㅎㅎ

[bm-common] [nio-8080-exec-9] reactor.Flux.MonoFlatMapMany.5           : onSubscribe(MonoFlatMapMany.FlatMapManyMain)
[bm-common] [nio-8080-exec-9] reactor.Flux.MonoFlatMapMany.5           : request(unbounded)
[bm-common] [ctor-http-nio-6] reactor.Flux.MonoFlatMapMany.5           : onComplete()
[bm-common] [ctor-http-nio-6] o.b.p.controller.PerftestController      : Test completed! action=stop
[bm-common] [ctor-http-nio-5] reactor.Flux.MonoFlatMapMany.4           : onComplete()
[bm-common] [ctor-http-nio-5] o.b.p.controller.PerftestController      : Test completed! action=start