Bugfix: don't throw away the container class of parameterized types (#950)

Author: matyasberry

services-api/src/main/java/io/scalecube/services/Reflect.java

@@ -25,6 +25,7 @@
 import java.util.HashSet;
 import java.util.Map;
 import java.util.Set;
+import java.util.concurrent.Future;
 import java.util.stream.Collectors;
 import java.util.stream.Stream;
 import org.reactivestreams.Publisher;
@@ -51,8 +52,8 @@ public static Type parameterizedReturnType(Method method) {
     }
 
     Type type = method.getGenericReturnType();
-    if (type instanceof ParameterizedType) {
-      Type actualReturnType = ((ParameterizedType) type).getActualTypeArguments()[0];
+    if (type instanceof ParameterizedType pt) {
+      Type actualReturnType = isAsyncContainer(type) ? pt.getActualTypeArguments()[0] : pt;
 
       if (ServiceMessage.class.equals(actualReturnType)) {
         return Object.class;
@@ -92,17 +93,13 @@ public static boolean isReturnTypeServiceMessage(Method method) {
    * @param method in inspection
    * @return type of method parameter
    */
-  public static Class<?> requestType(Method method) {
+  public static Type requestType(Method method) {
     if (method.getParameterTypes().length > 0) {
       if (method.isAnnotationPresent(RequestType.class)) {
         return method.getAnnotation(RequestType.class).value();
       } else {
         if (method.getGenericParameterTypes()[0] instanceof ParameterizedType) {
-          try {
-            return Class.forName(parameterizedRequestType(method).getTypeName());
-          } catch (ClassNotFoundException e) {
-            return Object.class;
-          }
+          return parameterizedRequestType(method);
         } else if (ServiceMessage.class.equals(method.getParameterTypes()[0])) {
           return Object.class;
         } else {
@@ -146,8 +143,8 @@ public static boolean isRequestTypeServiceMessage(Method method) {
   public static Type parameterizedType(Object object) {
     if (object != null) {
       Type type = object.getClass().getGenericSuperclass();
-      if (type instanceof ParameterizedType) {
-        return ((ParameterizedType) type).getActualTypeArguments()[0];
+      if (type instanceof ParameterizedType pt) {
+        return isAsyncContainer(type) ? pt.getActualTypeArguments()[0] : pt;
       }
     }
     return Object.class;
@@ -161,14 +158,36 @@ public static Type parameterizedType(Object object) {
   public static Type parameterizedRequestType(Method method) {
     if (method != null && method.getGenericParameterTypes().length > 0) {
       Type type = method.getGenericParameterTypes()[0];
-      if (type instanceof ParameterizedType) {
-        return ((ParameterizedType) type).getActualTypeArguments()[0];
+      if (type instanceof ParameterizedType pt) {
+        return isAsyncContainer(type) ? pt.getActualTypeArguments()[0] : pt;
       }
     }
 
     return Object.class;
   }
 
+  private static boolean isAsyncContainer(Type type) {
+    Class<?> raw = toClass(type);
+
+    if (raw == null) {
+      return false;
+    }
+
+    return Publisher.class.isAssignableFrom(raw)
+        || Stream.class.isAssignableFrom(raw)
+        || Future.class.isAssignableFrom(raw);
+  }
+
+  public static Class<?> toClass(Type type) {
+    if (type instanceof ParameterizedType pt) {
+      return (Class<?>) pt.getRawType();
+    }
+    if (type instanceof Class<?> c) {
+      return c;
+    }
+    return null;
+  }
+
   /**
    * Extracts service name from service interface.
    *

services-api/src/main/java/io/scalecube/services/methods/MethodInfo.java

@@ -1,5 +1,6 @@
 package io.scalecube.services.methods;
 
+import static io.scalecube.services.Reflect.toClass;
 import static io.scalecube.services.api.DynamicQualifier.isDynamicQualifier;
 
 import io.scalecube.services.CommunicationMode;
@@ -23,7 +24,7 @@ public class MethodInfo {
   private final boolean isReturnTypeServiceMessage;
   private final CommunicationMode communicationMode;
   private final int parameterCount;
-  private final Class<?> requestType;
+  private final Type requestType;
   private final boolean isRequestTypeServiceMessage;
   private final Secured secured;
   private final Scheduler scheduler;
@@ -53,7 +54,7 @@ public MethodInfo(
       boolean isReturnTypeServiceMessage,
       CommunicationMode communicationMode,
       int parameterCount,
-      Class<?> requestType,
+      Type requestType,
       boolean isRequestTypeServiceMessage,
       Secured secured,
       Scheduler scheduler,
@@ -115,10 +116,10 @@ public boolean isRequestTypeServiceMessage() {
   }
 
   public boolean isRequestTypeVoid() {
-    return requestType.isAssignableFrom(Void.TYPE);
+    return Void.TYPE.isAssignableFrom(toClass(requestType));
   }
 
-  public Class<?> requestType() {
+  public Type requestType() {
     return requestType;
   }
 

services-api/src/main/java/io/scalecube/services/transport/api/ServiceMessageDataDecoder.java

@@ -1,6 +1,7 @@
 package io.scalecube.services.transport.api;
 
 import io.scalecube.services.api.ServiceMessage;
+import java.lang.reflect.Type;
 import java.util.ServiceLoader;
 import java.util.stream.StreamSupport;
 
@@ -11,7 +12,7 @@ public interface ServiceMessageDataDecoder {
           .findFirst()
           .orElse(null);
 
-  ServiceMessage decodeData(ServiceMessage message, Class<?> dataType);
+  ServiceMessage decodeData(ServiceMessage message, Type type);
 
-  ServiceMessage copyData(ServiceMessage message, Class<?> dataType);
+  ServiceMessage copyData(ServiceMessage message, Type type);
 }

services-api/src/test/java/io/scalecube/services/methods/ServiceMethodInvokerTest.java

@@ -17,6 +17,7 @@
 import io.scalecube.services.exceptions.DefaultErrorMapper;
 import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
 import java.lang.reflect.Method;
+import java.lang.reflect.Type;
 import java.util.List;
 import java.util.function.Consumer;
 import java.util.stream.Stream;
@@ -39,12 +40,12 @@ class ServiceMethodInvokerTest {
   private final ServiceMessageDataDecoder dataDecoder =
       new ServiceMessageDataDecoder() {
         @Override
-        public ServiceMessage decodeData(ServiceMessage message, Class<?> dataType) {
+        public ServiceMessage decodeData(ServiceMessage message, Type type) {
           return message;
         }
 
         @Override
-        public ServiceMessage copyData(ServiceMessage message, Class<?> dataType) {
+        public ServiceMessage copyData(ServiceMessage message, Type type) {
           return message;
         }
       };

services-transport-parent/services-transport-rsocket/src/main/java/io/scalecube/services/transport/rsocket/ServiceMessageByteBufDataDecoder.java

@@ -2,16 +2,17 @@
 
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
+import java.lang.reflect.Type;
 
 public class ServiceMessageByteBufDataDecoder implements ServiceMessageDataDecoder {
 
   @Override
-  public ServiceMessage decodeData(ServiceMessage message, Class<?> dataType) {
+  public ServiceMessage decodeData(ServiceMessage message, Type dataType) {
     return ServiceMessageCodec.decodeData(message, dataType, false);
   }
 
   @Override
-  public ServiceMessage copyData(ServiceMessage message, Class<?> dataType) {
+  public ServiceMessage copyData(ServiceMessage message, Type dataType) {
     return ServiceMessageCodec.decodeData(message, dataType, true);
   }
 }

services/src/main/java/io/scalecube/services/Microservices.java

@@ -20,6 +20,7 @@
 import io.scalecube.services.transport.api.ServerTransport;
 import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
 import io.scalecube.services.transport.api.ServiceTransport;
+import java.lang.reflect.Type;
 import java.net.InetAddress;
 import java.net.UnknownHostException;
 import java.time.Duration;
@@ -766,12 +767,12 @@ private Context conclude() {
                 .orElse(
                     new ServiceMessageDataDecoder() {
                       @Override
-                      public ServiceMessage decodeData(ServiceMessage message, Class<?> dataType) {
+                      public ServiceMessage decodeData(ServiceMessage message, Type type) {
                         return message;
                       }
 
                       @Override
-                      public ServiceMessage copyData(ServiceMessage message, Class<?> dataType) {
+                      public ServiceMessage copyData(ServiceMessage message, Type type) {
                         return message;
                       }
                     });

services/src/test/java/io/scalecube/services/ServiceLocalTest.java

@@ -10,6 +10,8 @@
 import io.scalecube.services.sut.GreetingRequest;
 import io.scalecube.services.sut.GreetingResponse;
 import io.scalecube.services.sut.GreetingService;
+import io.scalecube.services.sut.GreetingService.Base;
+import io.scalecube.services.sut.GreetingService.MyPojo;
 import io.scalecube.services.sut.GreetingServiceImpl;
 import java.time.Duration;
 import java.util.concurrent.atomic.AtomicReference;
@@ -395,4 +397,19 @@ public void test_dynamic_qualifier() {
   private static GreetingService createProxy(Microservices gateway) {
     return gateway.call().api(GreetingService.class); // create proxy for GreetingService API
   }
+
+  @Test
+  public void test_generics_in_request() {
+    GreetingService service = createProxy(microservices);
+
+    final var pojo = new MyPojo("Joe", "NY");
+    // call the service.
+    final Mono<MyPojo> response =
+        service.greetingsWithGenerics(new Base<MyPojo>().object(pojo).format("plain"));
+    var result = response.block(TIMEOUT.plusMillis(500));
+
+    // print the greeting.
+    System.out.println("test_generics_in_request : " + result);
+    assertEquals(pojo, result);
+  }
 }

services/src/test/java/io/scalecube/services/ServiceRemoteTest.java

@@ -17,6 +17,7 @@
 import io.scalecube.services.sut.GreetingRequest;
 import io.scalecube.services.sut.GreetingResponse;
 import io.scalecube.services.sut.GreetingService;
+import io.scalecube.services.sut.GreetingService.MyPojo;
 import io.scalecube.services.sut.GreetingServiceImpl;
 import io.scalecube.services.transport.rsocket.RSocketServiceTransport;
 import io.scalecube.transport.netty.websocket.WebsocketTransportFactory;
@@ -572,6 +573,17 @@ public void test_dynamic_qualifier() {
         .verifyComplete();
   }
 
+  @Test
+  public void test_generics_in_request() {
+    final var service = gateway.call().api(GreetingService.class);
+    final var pojo = new GreetingService.MyPojo("Joe", "NY");
+
+    StepVerifier.create(
+            service.greetingsWithGenerics(new GreetingService.Base<MyPojo>().object(pojo)))
+        .assertNext(result -> assertEquals(pojo, result))
+        .verifyComplete();
+  }
+
   private GreetingService createProxy() {
     return gateway.call().api(GreetingService.class); // create proxy for GreetingService API
   }

services/src/test/java/io/scalecube/services/sut/GreetingService.java

@@ -103,4 +103,24 @@ Flux<ServiceMessage> bidiGreetingIllegalArgumentExceptionMessage(
 
   @ServiceMethod("hello/:someVar")
   Mono<String> helloDynamicQualifier(Long value);
+
+  public static class Base<T> {
+    T obj;
+    String format;
+
+    public Base object(T obj) {
+      this.obj = obj;
+      return this;
+    }
+
+    public Base format(String format) {
+      this.format = format;
+      return this;
+    }
+  }
+
+  @ServiceMethod
+  Mono<MyPojo> greetingsWithGenerics(Base<MyPojo> request);
+
+  public record MyPojo(String name, String address) {}
 }

services/src/test/java/io/scalecube/services/sut/GreetingServiceImpl.java

@@ -207,4 +207,9 @@ public Mono<String> helloDynamicQualifier(Long value) {
     return RequestContext.deferContextual()
         .map(context -> context.pathParams().getString("someVar") + "@" + value);
   }
+
+  @Override
+  public Mono<MyPojo> greetingsWithGenerics(Base<MyPojo> request) {
+    return Mono.just(request.obj);
+  }
 }