gistfile1.txt

    @Test
    void should_fail_when_listener_onOpen_throws_exception() throws Exception {

        for (HttpClient client : clients()) {

            // given
            HttpRequest request = HttpRequest.builder()
                    .method(POST)
                    .url("https://api.openai.com/v1/chat/completions")
                    .addHeader("Authorization", "Bearer " + OPENAI_API_KEY)
                    .addHeader("Content-Type", "application/json")
                    .body("""
                            {
                                "model": "gpt-4o-mini",
                                "messages": [
                                    {
                                        "role" : "user",
                                        "content" : "What is the capital of Germany?"
                                    }
                                ],
                                "stream": true
                            }
                            """)
                    .build();

            // when
            AtomicReference<SuccessfulHttpResponse> response = new AtomicReference<>();
            List<ServerSentEvent> events = synchronizedList(new ArrayList<>());
            List<Throwable> errors = synchronizedList(new ArrayList<>());
            Set<Thread> threads = synchronizedSet(new HashSet<>());

            CountDownLatch doneSignal = new CountDownLatch(1);

            ServerSentEventListener listener = new ServerSentEventListener() {

                @Override
                public void onOpen(SuccessfulHttpResponse successfulHttpResponse) {
                    response.set(successfulHttpResponse);
                    threads.add(Thread.currentThread());

                    throw new RuntimeException("Unexpected exception in onOpen()");
                }

                @Override
                public void onEvent(ServerSentEvent event) {
                    events.add(event);
                    threads.add(Thread.currentThread());
                }

                @Override
                public void onError(Throwable throwable) {
                    errors.add(throwable);
                    threads.add(Thread.currentThread());
                }

                @Override
                public void onClose() {
                    threads.add(Thread.currentThread());
                    doneSignal.countDown();
                }
            };
            ServerSentEventListener spyListener = spy(listener);
            client.execute(request, new DefaultServerSentEventParser(), spyListener);
            doneSignal.await();

            // then
            assertThat(response.get()).isNotNull();
            assertThat(events).isEmpty();
            assertThat(errors).isEmpty();

            assertThat(threads).hasSize(1);
            assertThat(threads.iterator().next()).isNotEqualTo(Thread.currentThread());

            InOrder inOrder = inOrder(spyListener);
            inOrder.verify(spyListener, times(1)).onOpen(any());
            inOrder.verifyNoMoreInteractions();
        }
    }