package io.scalecube.services.methods;

import io.scalecube.services.api.ServiceMessage;
import io.scalecube.services.exceptions.BadRequestException;
import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
import org.reactivestreams.Publisher;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import java.lang.reflect.InvocationTargetException;
import java.lang.reflect.Method;
import java.lang.reflect.Parameter;
import java.util.Optional;
import java.util.stream.Stream;

import static java.util.stream.Collectors.joining;

/**
 * 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;
  private final ServiceMessageDataDecoder dataDecoder;

  /**
   * 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
   * @param dataDecoder data decoder
   */
  public ServiceMethodInvoker(
      Method method,
      Object service,
      MethodInfo methodInfo,
      ServiceProviderErrorMapper errorMapper,
      ServiceMessageDataDecoder dataDecoder) {
    this.method = method;
    this.service = service;
    this.methodInfo = methodInfo;
    this.errorMapper = errorMapper;
    this.dataDecoder = dataDecoder;
  }

  /**
   * Invokes service method with single response.
   *
   * @param message request service message
   * @return mono of service message
   */
  public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
    return Mono.defer(() -> Mono.from(invoke(toRequest(message))))
        .map(this::toResponse)
        .onErrorResume(throwable -> Mono.just(errorMapper.toMessage(throwable)));
  }

  /**
   * Invokes service method with message stream response.
   *
   * @param message request service message
   * @return flux of service messages
   */
  public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
    return Flux.defer(() -> Flux.from(invoke(toRequest(message))))
        .map(this::toResponse)
        .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
  }

  /**
   * Invokes service method with bidirectional communication.
   *
   * @param publisher request service message
   * @return flux of service messages
   */
  public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publisher) {
    return Flux.from(publisher)
        .map(this::toRequest)
        .transform(this::invoke)
        .map(this::toResponse)
        .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) {
    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();
  }

  @Override
  public String toString() {
    String prefix = service.getClass().getCanonicalName() + "#" + method.getName() + "(";
    return Stream.of(method.getParameters())
        .map(Parameter::getType)
        .map(Class::getSimpleName)
        .collect(joining(", ", prefix, ")"));
  }
}

@@ -285,119 +285,8 @@
 er;%0A
-import org.reactivestreams.Publisher;%0Aimport reactor.core.publisher.Flux;%0Aimport reactor.core.publisher.Mono;%0A%0A
 impo
@@ -429,16 +429,52 @@
 tional;%0A
+import java.util.stream.Collectors;%0A
 import j
@@ -501,58 +501,116 @@
 am;%0A
-%0A
 import 
-static java.util.stream.Collectors.joining
+org.reactivestreams.Publisher;%0Aimport reactor.core.publisher.Flux;%0Aimport reactor.core.publisher.Mono
 ;%0A%0A/
@@ -4564,22 +4564,30 @@
  String 
-prefix
+classAndMethod
  = servi
@@ -4647,26 +4647,27 @@
 me()
- + %22(%22;%0A    return
+;%0A    String args =
  Str
@@ -4701,24 +4701,28 @@
 ())%0A        
+  
+  
 .map(Paramet
@@ -4734,16 +4734,20 @@
 etType)%0A
+    
         
@@ -4781,16 +4781,20 @@
         
+ 
+   
 .collect
@@ -4794,16 +4794,27 @@
 collect(
+Collectors.
 joining(
@@ -4823,14 +4823,11 @@
  %22, 
-prefix
+%22(%22
 , %22)
@@ -4827,18 +4827,52 @@
 %22(%22, %22)%22));%0A
+    return classAndMethod + args;%0A
   %7D%0A%7D%0A

fix codestyle