Repository: apache/dubbo
File: dubbo-plugin/dubbo-mutiny/src/test/java/org/apache/dubbo/mutiny/MutinyClientCallsTest.java (Java)

Code before:
            });
            Assertions.assertTrue(ex.getMessage().contains("boom"));
        }
    }

    @Test
    void testOneToManyReturnsMultiAndEmitsItems() {
        Invoker<Object> invoker = Mockito.mock(Invoker.class);
        StubMethodDescriptor method = Mockito.mock(StubMethodDescriptor.class);

        try (MockedStatic<StubInvocationUtil> mocked = Mockito.mockStatic(StubInvocationUtil.class)) {
            AtomicBoolean stubCalled = new AtomicBoolean(false);

            mocked.when(() -> StubInvocationUtil.serverStreamCall(
                            Mockito.eq(invoker), Mockito.eq(method), Mockito.eq("testRequest"), Mockito.any()))
                    .thenAnswer(invocation -> {
                        stubCalled.set(true);
                        ClientTripleMutinyPublisher<String> publisher = invocation.getArgument(3);

                        CallStreamObserver<String> fakeSubscription = new CallStreamObserver<>() {
                            @Override
                            public void request(int n) {}

                            @Override
                            public void setCompression(String compression) {}

                            @Override
                            public void disableAutoFlowControl() {}

                            @Override
                            public void onNext(String value) {
                                publisher.onNext(value);
                            }

                            @Override
                            public void onError(Throwable t) {
                                publisher.onError(t);
                            }

                            @Override
                            public void onCompleted() {
                                publisher.onCompleted();
                            }
                        };

                        publisher.onSubscribe(fakeSubscription);

                        new Thread(() -> {
                                    publisher.onNext("item1");
                                    publisher.onNext("item2");
                                    publisher.onCompleted();

Review comment (suggestion):
[nitpick] The comment '/* no-op */' should be more descriptive to explain why this method is intentionally empty in the test context.
```suggestion
                                // Intentionally left empty: flow control is not required for this test's mock observer.
```

Code after:
            RuntimeException ex = Assertions.assertThrows(RuntimeException.class, () -> {
                response.await().indefinitely();
            });
            Assertions.assertTrue(ex.getMessage().contains("boom"));
        }
    }

    @Test
    void testOneToManyReturnsMultiAndEmitsItems() {
        Invoker<Object> invoker = Mockito.mock(Invoker.class);
        StubMethodDescriptor method = Mockito.mock(StubMethodDescriptor.class);

        try (MockedStatic<StubInvocationUtil> mocked = Mockito.mockStatic(StubInvocationUtil.class)) {
            AtomicBoolean stubCalled = new AtomicBoolean(false);
            CountDownLatch subscribed = new CountDownLatch(1);

            mocked.when(() -> StubInvocationUtil.serverStreamCall(
                            Mockito.eq(invoker), Mockito.eq(method), Mockito.eq("testRequest"), Mockito.any()))
                    .thenAnswer(invocation -> {
                        stubCalled.set(true);
                        ClientTripleMutinyPublisher<String> publisher = invocation.getArgument(3);

                        CallStreamObserver<String> fakeSubscription = new CallStreamObserver<>() {
                            @Override
                            public void request(int n) {
                                /* no-op */
                            }

                            @Override
                            public void setCompression(String compression) {}

                            @Override
                            public void disableAutoFlowControl() {}

                            @Override
                            public void onNext(String v) {
                                publisher.onNext(v);
                            }

                            @Override
                            public void onError(Throwable t) {
                                publisher.onError(t);
                            }

                            @Override
                            public void onCompleted() {
                                publisher.onCompleted();
                            }
                        };
                        publisher.onSubscribe(fakeSubscription);