package io.scalecube.services;

import io.scalecube.services.api.ServiceMessage;

import java.lang.reflect.Method;
import java.lang.reflect.Type;

import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

public class LocalServiceMethodInvoke implements ServiceMethodInvoke {

  Method method;
  Object serviceObject;
  Type requestType = Reflect.parameterizedType(method);
  Type returnType = Reflect.parameterizedReturnType(method);

  @Override
  public Mono<ServiceMessage> requestResponse(ServiceMessage request) {
    
    Object object = toObject(decodeData(request, requestType));
    return invokeReturnsMono(object)
        .map(val -> toMessage(val))
        .map(resp -> encodeData(resp));
  }

  @Override
  public Mono<Void> fireAndForget(ServiceMessage request) {
    Object object = toObject(decodeData(request, requestType));
    invokeReturnsMono(object);
    return Mono.empty();
  }


  @Override
  public Flux<ServiceMessage> requestChannel(Flux<ServiceMessage> request) {
    return request.map(req -> decodeData(req, requestType))
        .map(message -> toObject(message))
        .map(object -> invokeReturnsFlux(object))
        .map(object -> toMessage(object))
        .map(resp -> encodeData(resp));
  }

  @Override
  public Flux<ServiceMessage> requestStream(ServiceMessage request) {
    Object object = toObject(decodeData(request, requestType));
    return invokeReturnsFlux(object)
        .map(val -> toMessage(val))
        .map(resp -> encodeData(resp));
  }

  private <T> Flux<T> invokeReturnsFlux(Object object) {
    try {
      return (Flux<T>) method.invoke(serviceObject, object);
    } catch (Exception ex) {
      ex.printStackTrace();
    }
    return null;
  }

  private <T> Mono<T> invokeReturnsMono(Object object) {
    try {
      return (Mono<T>) method.invoke(serviceObject, object);
    } catch (Exception ex) {
      ex.printStackTrace();
    }
    return null;
  }

  private ServiceMessage decodeData(ServiceMessage msg, Type requestType) {
    // TODO Auto-generated method stub
    return null;
  }

  private ServiceMessage encodeData(ServiceMessage resp) {
    // TODO Auto-generated method stub
    return null;
  }

  private ServiceMessage toMessage(Object object) {
    if (requestType.equals(ServiceMessage.class)) {
      return (ServiceMessage) object;
    } else {
      return ServiceMessage.builder().data(object).build();
    }
  }

  private Object toObject(ServiceMessage message) {
    if (returnType.equals(ServiceMessage.class)) {
      return message;
    } else {
      return message.data();
    }
  }
}

@@ -536,37 +536,37 @@
     %0A    Object 
-objec
+resul
 t = toObject(dec
@@ -626,21 +626,21 @@
 rnsMono(
-objec
+resul
 t)%0A     
@@ -639,35 +639,38 @@
 t)%0A        .map(
-val
+object
  -%3E toMessage(va
@@ -659,35 +659,38 @@
 ct -%3E toMessage(
-val
+object
 ))%0A        .map(
@@ -861,16 +861,23 @@
 ));%0A    
+return 
 invokeRe
@@ -899,33 +899,8 @@
 t);%0A
-    return Mono.empty();%0A
   %7D%0A
@@ -1315,37 +1315,37 @@
 t) %7B%0A    Object 
-objec
+resul
 t = toObject(dec
@@ -1405,21 +1405,21 @@
 rnsFlux(
-objec
+resul
 t)%0A     
@@ -1426,19 +1426,22 @@
    .map(
-val
+object
  -%3E toMe
@@ -1450,11 +1450,14 @@
 age(
-val
+object
 ))%0A 
@@ -1667,52 +1667,36 @@
     
-ex.printStackTrace();%0A    %7D%0A    return null;
+return Flux.error(ex);%0A    %7D
 %0A  %7D
@@ -1864,52 +1864,36 @@
     
-ex.printStackTrace();%0A    %7D%0A    return null;
+return Mono.error(ex);%0A    %7D
 %0A  %7D

return error