package io.scalecube.services.methods;

import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.exceptions.BadRequestException;
import io.scalecube.services.exceptions.mappers.ServiceProviderErrorMapper;
import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.util.Optional;
import java.util.function.BiFunction;
import java.util.function.Supplier;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

/**
 * Invoker of service method. Prepares service message request before call as well as doing some
 * handling of a product of the service call.
 */
public final class ServiceMethodInvoker {

  private final Method method;
  private final Object service;
  private final MethodInfo methodInfo;
  private final ServiceProviderErrorMapper errorMapper;

  /**
   * Constructs a service method invoker out of real service object instance and method info.
   *
   * @param method service method
   * @param service service instance
   * @param methodInfo method information
   * @param errorMapper error mapper
   */
  public ServiceMethodInvoker(
      Method method,
      Object service,
      MethodInfo methodInfo,
      ServiceProviderErrorMapper errorMapper) {
    this.method = method;
    this.service = service;
    this.methodInfo = methodInfo;
    this.errorMapper = errorMapper;
  }

  /**
   * Invokes service method with single response.
   *
   * @param message request service message
   * @param dataDecoder function to create new service message with decoded data
   * @return mono of service message
   */
  public Mono<ServiceMessage> invokeOne(
      ServiceMessage message, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder) {
    Supplier<Mono<ServiceMessage>> invoke =
        () -> {
          Object request = toRequest(message, dataDecoder);
          return Mono.from(invoke(request)).map(this::toResponse);
        };
    return Mono.defer(invoke)
        .onErrorResume(throwable -> Mono.just(errorMapper.toMessage(throwable)));
  }

  /**
   * Invokes service method with message stream response.
   *
   * @param message request service message
   * @param dataDecoder function to create new service message with decoded data
   * @return flux of service messages
   */
  public Flux<ServiceMessage> invokeMany(
      ServiceMessage message, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder) {
    Supplier<Publisher<ServiceMessage>> invoke =
        () -> {
          Object request = toRequest(message, dataDecoder);
          return Flux.from(invoke(request)).map(this::toResponse);
        };
    return Flux.defer(invoke)
        .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
  }

  /**
   * Invokes service method with bidirectional communication.
   *
   * @param publisher request service message
   * @param dataDecoder function to create new service message with decoded data
   * @return flux of service messages
   */
  public Flux<ServiceMessage> invokeBidirectional(
      Publisher<ServiceMessage> publisher,
      BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder) {
    Supplier<Publisher<ServiceMessage>> invoke =
        () -> {
          Flux<?> requestPublisher =
              Flux.from(publisher).map(message -> toRequest(message, dataDecoder));
          return Flux.from(invoke(requestPublisher)).map(this::toResponse);
        };
    return Flux.defer(invoke)
        .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
  }

  private Publisher<?> invoke(Object arguments) {
    Publisher<?> result = null;
    Throwable throwable = null;
    try {
      if (method.getParameterCount() == 0) {
        result = (Publisher<?>) method.invoke(service);
      } else {
        result = (Publisher<?>) method.invoke(service, arguments);
      }
      if (result == null) {
        result = Mono.empty();
      }
    } catch (InvocationTargetException ex) {
      throwable = Optional.ofNullable(ex.getCause()).orElse(ex);
    } catch (Throwable ex) {
      throwable = ex;
    }
    return throwable != null ? Mono.error(throwable) : result;
  }

  private Object toRequest(
      ServiceMessage message, BiFunction<ServiceMessage, Class<?>, ServiceMessage> dataDecoder) {
    ServiceMessage request = dataDecoder.apply(message, methodInfo.requestType());

    if (!methodInfo.isRequestTypeVoid()
        && !methodInfo.isRequestTypeServiceMessage()
        && !request.hasData(methodInfo.requestType())) {

      Optional<?> dataOptional = Optional.ofNullable(request.data());
      Class<?> clazz = dataOptional.map(Object::getClass).orElse(null);
      throw new BadRequestException(
          String.format(
              "Expected service request data of type: %s, but received: %s",
              methodInfo.requestType(), clazz));
    }

    return methodInfo.isRequestTypeServiceMessage() ? request : request.data();
  }

  private ServiceMessage toResponse(Object response) {
    return (response instanceof ServiceMessage)
        ? (ServiceMessage) response
        : ServiceMessage.builder().qualifier(methodInfo.qualifier()).data(response).build();
  }
}

@@ -373,44 +373,8 @@
 on;%0A
-import java.util.function.Supplier;%0A
 impo
@@ -403,24 +403,24 @@
 .Publisher;%0A
+
 import react
@@ -1750,91 +1750,49 @@
     
-Supplier%3CMono%3CServiceMessage%3E%3E invoke =%0A        () -%3E %7B%0A          Object request = 
+return Mono.defer(() -%3E Mono.from(invoke(
 toRe
@@ -1810,33 +1810,35 @@
 ge, dataDecoder)
-;
+)))
 %0A          retur
@@ -1834,105 +1834,28 @@
     
-  return Mono.from(invoke(request)).map(this::toResponse);%0A        %7D;%0A    return Mono.defer(invok
+.map(this::toRespons
 e)%0A 
@@ -2326,96 +2326,49 @@
     
-Supplier%3CPublisher%3CServiceMessage%3E%3E invoke =%0A        () -%3E %7B%0A          Object request = 
+return Flux.defer(() -%3E Flux.from(invoke(
 toRe
@@ -2394,17 +2394,19 @@
 Decoder)
-;
+)))
 %0A       
@@ -2410,105 +2410,28 @@
     
-  return Flux.from(invoke(request)).map(this::toResponse);%0A        %7D;%0A    return Flux.defer(invok
+.map(this::toRespons
 e)%0A 
@@ -2936,103 +2936,34 @@
     
-Supplier%3CPublisher%3CServiceMessage%3E%3E invoke =%0A        () -%3E %7B%0A          Flux%3C?%3E requestP
+return Flux.from(p
 ublisher
  =%0A 
@@ -2962,22 +2962,18 @@
 sher
- =%0A   
+)%0A
         
    F
@@ -2972,31 +2972,8 @@
     
-   Flux.from(publisher)
 .map
@@ -3016,17 +3016,16 @@
 ecoder))
-;
 %0A       
@@ -3029,114 +3029,61 @@
     
-  return Flux.from(invoke(requestPublisher)).map(this::toResponse);%0A        %7D;%0A    return Flux.defer(invok
+.transform(this::invoke)%0A        .map(this::toRespons
 e)%0A 

Refactor code to get rid of `Supplier` variable