### Question:
ServiceMethodInvoker { public Flux<ServiceMessage> invokeBidirectional( Publisher<ServiceMessage> publisher, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder) { return Flux.from(invoke(Flux.from(publisher).map(message -> toRequest(message, dataDecoder)))) .map(this::toResponse); } ServiceMethodInvoker(Method method, Object service, MethodInfo methodInfo); Mono<ServiceMessage> invokeOne(
      ServiceMessage message, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder); Flux<ServiceMessage> invokeMany(
      ServiceMessage message, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder); Flux<ServiceMessage> invokeBidirectional(
      Publisher<ServiceMessage> publisher,
      BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder);  }### Answer:
@Test @DisplayName("invokeBidirectional should return empty response when service returns null") void testInvokeBidirectionalWhenReturnNull() throws Exception { final String methodName = "returnNull3"; final Class<? extends StubService> serviceClass = stubService.getClass(); final Method method = serviceClass.getMethod(methodName, Flux.class); final MethodInfo methodInfo = new MethodInfo( serviceClass.getName(), methodName, method.getReturnType(), CommunicationMode.REQUEST_CHANNEL, method.getParameterCount(), Void.TYPE); serviceMethodInvoker = new ServiceMethodInvoker(method, stubService, methodInfo); ServiceMessage message = ServiceMessage.builder().qualifier(qualifierPrefix + methodName).streamId("1").build(); StepVerifier.create(serviceMethodInvoker.invokeBidirectional(Flux.just(message), dataDecoder)) .verifyComplete(); }
@Test @DisplayName("invokeBidirectional should return error response when service throws exception") void testInvokeBidirectionalWhenThrowException() throws Exception { final String methodName = "throwException3"; final Class<? extends StubService> serviceClass = stubService.getClass(); final Method method = serviceClass.getMethod(methodName, Flux.class); final MethodInfo methodInfo = new MethodInfo( serviceClass.getName(), methodName, method.getReturnType(), CommunicationMode.REQUEST_CHANNEL, method.getParameterCount(), Void.TYPE); serviceMethodInvoker = new ServiceMethodInvoker(method, stubService, methodInfo); ServiceMessage message = ServiceMessage.builder().qualifier(qualifierPrefix + methodName).streamId("1").build(); final Flux<ServiceMessage> invokeOne = serviceMethodInvoker.invokeBidirectional(Flux.just(message), dataDecoder); StepVerifier.create(invokeOne).verifyError(RuntimeException.class); }