@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(), IS_RETURN_TYPE_SERVICE_MESSAGE, CommunicationMode.REQUEST_CHANNEL, method.getParameterCount(), Void.TYPE, IS_REQUEST_TYPE_SERVICE_MESSAGE, AUTH); serviceMethodInvoker = new ServiceMethodInvoker( method, stubService, methodInfo, DefaultErrorMapper.INSTANCE, dataDecoder, nullAuthenticator, nullPrincipalMapper); ServiceMessage message = ServiceMessage.builder().qualifier(qualifierPrefix + methodName).build(); final Flux<ServiceMessage> invokeOne = Flux.deferWithContext( context -> serviceMethodInvoker.invokeBidirectional(Flux.just(message))) .subscriberContext(context -> context.put(AUTH_CONTEXT_KEY, AUTH_DATA)); StepVerifier.create(invokeOne) .assertNext(serviceMessage -> Assertions.assertTrue(serviceMessage.isError())) .verifyComplete(); }
public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) { return Flux.from(publisher) .switchOnFirst( (first, messages) -> Mono.deferWithContext(context -> authenticate(first.get(), context)) .flatMapMany(authData -> deferWithContextBidirectional(messages, authData)) .map( response -> toResponse(response, first.get().qualifier(), first.get().dataFormat())) .onErrorResume( throwable -> Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))); }
ServiceMethodInvoker { public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) { return Flux.from(publisher) .switchOnFirst( (first, messages) -> Mono.deferWithContext(context -> authenticate(first.get(), context)) .flatMapMany(authData -> deferWithContextBidirectional(messages, authData)) .map( response -> toResponse(response, first.get().qualifier(), first.get().dataFormat())) .onErrorResume( throwable -> Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))); } }
ServiceMethodInvoker { public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) { return Flux.from(publisher) .switchOnFirst( (first, messages) -> Mono.deferWithContext(context -> authenticate(first.get(), context)) .flatMapMany(authData -> deferWithContextBidirectional(messages, authData)) .map( response -> toResponse(response, first.get().qualifier(), first.get().dataFormat())) .onErrorResume( throwable -> Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))); } ServiceMethodInvoker(
      Method method,
      Object service,
      MethodInfo methodInfo,
      ServiceProviderErrorMapper errorMapper,
      ServiceMessageDataDecoder dataDecoder,
      Authenticator<Object> authenticator,
      PrincipalMapper<Object, Object> principalMapper); }
ServiceMethodInvoker { public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) { return Flux.from(publisher) .switchOnFirst( (first, messages) -> Mono.deferWithContext(context -> authenticate(first.get(), context)) .flatMapMany(authData -> deferWithContextBidirectional(messages, authData)) .map( response -> toResponse(response, first.get().qualifier(), first.get().dataFormat())) .onErrorResume( throwable -> Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))); } ServiceMethodInvoker(
      Method method,
      Object service,
      MethodInfo methodInfo,
      ServiceProviderErrorMapper errorMapper,
      ServiceMessageDataDecoder dataDecoder,
      Authenticator<Object> authenticator,
      PrincipalMapper<Object, Object> principalMapper); Mono<ServiceMessage> invokeOne(ServiceMessage message); Flux<ServiceMessage> invokeMany(ServiceMessage message); Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher); Object service(); MethodInfo methodInfo(); @Override String toString(); }
ServiceMethodInvoker { public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) { return Flux.from(publisher) .switchOnFirst( (first, messages) -> Mono.deferWithContext(context -> authenticate(first.get(), context)) .flatMapMany(authData -> deferWithContextBidirectional(messages, authData)) .map( response -> toResponse(response, first.get().qualifier(), first.get().dataFormat())) .onErrorResume( throwable -> Flux.just(errorMapper.toMessage(first.get().qualifier(), throwable)))); } ServiceMethodInvoker(
      Method method,
      Object service,
      MethodInfo methodInfo,
      ServiceProviderErrorMapper errorMapper,
      ServiceMessageDataDecoder dataDecoder,
      Authenticator<Object> authenticator,
      PrincipalMapper<Object, Object> principalMapper); Mono<ServiceMessage> invokeOne(ServiceMessage message); Flux<ServiceMessage> invokeMany(ServiceMessage message); Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher); Object service(); MethodInfo methodInfo(); @Override String toString();  }