Merge pull request #791 from harry-hao/feature/generic-is-request-type-service-message

segabriel · web-flow · commit cc83cd96bd30 · 2020-08-07T21:27:16.000+03:00
Fixed an error caused by Reflect.isRequestTypeServiceMessage() returning false on Publisher&lt;ServiceMessage&gt;
diff --git a/services-api/src/main/java/io/scalecube/services/Reflect.java b/services-api/src/main/java/io/scalecube/services/Reflect.java
@@ -115,9 +115,18 @@ public static Class<?> requestType(Method method) {
    * @return true if the first parameter of method is ServiceMessage, otherwise false
    */
   public static boolean isRequestTypeServiceMessage(Method method) {
-    Class<?>[] parameterTypes = method.getParameterTypes();
+    Type[] parameterTypes = method.getGenericParameterTypes();
 
-    return parameterTypes.length > 0 && ServiceMessage.class.equals(parameterTypes[0]);
+    if (parameterTypes.length < 1) {
+      return false;
+    }
+
+    if (parameterTypes[0] instanceof ParameterizedType) {
+      ParameterizedType parameterizedType = (ParameterizedType) parameterTypes[0];
+      return ServiceMessage.class.equals(parameterizedType.getActualTypeArguments()[0]);
+    }
+
+    return ServiceMessage.class.equals(parameterTypes[0]);
   }
 
   /**
diff --git a/services-api/src/test/java/io/scalecube/services/methods/ReflectTest.java b/services-api/src/test/java/io/scalecube/services/methods/ReflectTest.java
@@ -10,6 +10,8 @@
 import java.lang.reflect.Method;
 import java.util.Arrays;
 import java.util.stream.Stream;
+
+import io.scalecube.services.api.ServiceMessage;
 import org.junit.jupiter.api.Assertions;
 import org.junit.jupiter.params.ParameterizedTest;
 import org.junit.jupiter.params.provider.Arguments;
@@ -26,7 +28,7 @@ public class ReflectTest {
    * @param expectedMode expected mode
    */
   @ParameterizedTest
-  @MethodSource("argsProvider")
+  @MethodSource("argsCommunicationModeProvider")
   public void testCommunicationMode(String methodName, CommunicationMode expectedMode) {
     // Given:
     Method m =
@@ -40,12 +42,47 @@ public void testCommunicationMode(String methodName, CommunicationMode expectedM
     Assertions.assertEquals(expectedMode, communicationMode, "Invalid communicationMode");
   }
 
-  static Stream<Arguments> argsProvider() {
+  static Stream<Arguments> argsCommunicationModeProvider() {
     return Stream.of(
         Arguments.of("fireAndForget", FIRE_AND_FORGET),
+        Arguments.of("emptyResponse", REQUEST_RESPONSE),
         Arguments.of("requestResponse", REQUEST_RESPONSE),
         Arguments.of("requestStream", REQUEST_STREAM),
-        Arguments.of("requestChannel", REQUEST_CHANNEL));
+        Arguments.of("requestChannel", REQUEST_CHANNEL),
+        Arguments.of("fireAndForgetMessage", FIRE_AND_FORGET),
+        Arguments.of("emptyResponseMessage", REQUEST_RESPONSE),
+        Arguments.of("requestResponseMessage", REQUEST_RESPONSE),
+        Arguments.of("requestStreamMessage", REQUEST_STREAM),
+        Arguments.of("requestChannelMessage", REQUEST_CHANNEL));
+  }
+
+  @ParameterizedTest
+  @MethodSource("argsIsRequestTypeServiceMessage")
+  public void testIsRequestTypeServiceMessage(String methodName, boolean expect) {
+    // Given:
+    Method method = Arrays.stream(ReflectTest.TestService.class.getMethods())
+            .filter(meth -> meth.getName().equals(methodName))
+            .findFirst()
+            .get();
+    // When:
+    boolean actual = Reflect.isRequestTypeServiceMessage(method);
+    // Then:
+    Assertions.assertEquals(expect, actual,
+            String.format("isRequestTypeServiceMessage(%s) should be %b", methodName, expect));
+  }
+
+  static Stream<Arguments> argsIsRequestTypeServiceMessage() {
+    return Stream.of(
+            Arguments.of("fireAndForget", false),
+            Arguments.of("emptyResponse", false),
+            Arguments.of("requestResponse", false),
+            Arguments.of("requestStream", false),
+            Arguments.of("requestChannel", false),
+            Arguments.of("fireAndForgetMessage", true),
+            Arguments.of("emptyResponseMessage", true),
+            Arguments.of("requestResponseMessage", true),
+            Arguments.of("requestStreamMessage", true),
+            Arguments.of("requestChannelMessage", true));
   }
 
   private interface TestService {
@@ -58,5 +95,16 @@ private interface TestService {
     Flux<Integer> requestStream(Integer i);
 
     Flux<Integer> requestChannel(Flux<Integer> i);
+
+    void fireAndForgetMessage(ServiceMessage sm);
+
+    Mono<Void> emptyResponseMessage(ServiceMessage sm);
+
+    Mono<ServiceMessage> requestResponseMessage(ServiceMessage sm);
+
+    Flux<ServiceMessage> requestStreamMessage(ServiceMessage sm);
+
+    Flux<ServiceMessage> requestChannelMessage(Flux<ServiceMessage> sm);
+
   }
 }
diff --git a/services/src/test/java/io/scalecube/services/ServiceLocalTest.java b/services/src/test/java/io/scalecube/services/ServiceLocalTest.java
@@ -14,9 +14,7 @@
 import org.junit.jupiter.api.AfterEach;
 import org.junit.jupiter.api.BeforeEach;
 import org.junit.jupiter.api.Test;
-import reactor.core.publisher.EmitterProcessor;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.*;
 import reactor.test.StepVerifier;
 
 public class ServiceLocalTest extends BaseTest {
@@ -260,6 +258,23 @@ public void test_local_bidi_greeting_expect_IllegalArgumentException() {
         .verify(Duration.ofSeconds(3));
   }
 
+    @Test
+    public void test_local_bidi_greeting_message_expect_IllegalArgumentException() {
+        // get a proxy to the service api.
+        GreetingService service = createProxy(microservices);
+
+        // call the service. bidiThrowingGreeting
+        Flux<GreetingResponse> responses =
+                service.bidiGreetingIllegalArgumentExceptionMessage(
+                        Mono.just(ServiceMessage.builder()
+                                .data(new GreetingRequest("IllegalArgumentException")).build()))
+                .map(ServiceMessage::data);
+        // call the service.
+        StepVerifier.create(responses)
+                .expectErrorMessage("IllegalArgumentException")
+                .verify(Duration.ofSeconds(3));
+    }
+
   @Test
   public void test_local_bidi_greeting_expect_NotAuthorized() {
     // get a proxy to the service api.
@@ -279,6 +294,24 @@ public void test_local_bidi_greeting_expect_NotAuthorized() {
         .verify(Duration.ofSeconds(3));
   }
 
+    @Test
+    public void test_local_bidi_greeting_message_expect_NotAuthorized() {
+        // get a proxy to the service api.
+        GreetingService service = createProxy(microservices);
+
+        DirectProcessor<GreetingRequest> requests = DirectProcessor.create();
+
+        // call the service.
+        Flux<GreetingResponse> responses = service.bidiGreetingNotAuthorizedMessage(
+                requests.map(request -> ServiceMessage.builder().data(request).build()))
+                .map(ServiceMessage::data);
+
+        StepVerifier.create(responses)
+                .then(() -> requests.onNext(new GreetingRequest("joe-1")))
+                .expectErrorMessage("Not authorized")
+                .verify(Duration.ofSeconds(3));
+    }
+
   @Test
   public void test_local_bidi_greeting_expect_GreetingResponse() {
     // get a proxy to the service api.
@@ -304,6 +337,29 @@ public void test_local_bidi_greeting_expect_GreetingResponse() {
         .verify(Duration.ofSeconds(3));
   }
 
+    @Test
+    public void test_local_bidi_greeting_expect_message_GreetingResponse() {
+        // get a proxy to the service api.
+        GreetingService service = createProxy(microservices);
+
+        UnicastProcessor<GreetingRequest> requests = UnicastProcessor.create();
+        // call the service.
+        Flux<GreetingResponse> responses = service.bidiGreetingMessage(requests
+                .map(request -> ServiceMessage.builder().data(request).build()))
+                .map(ServiceMessage::data);
+
+        StepVerifier.create(responses)
+                .then(() -> requests.onNext(new GreetingRequest("joe-1")))
+                .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-1"))
+                .then(() -> requests.onNext(new GreetingRequest("joe-2")))
+                .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-2"))
+                .then(() -> requests.onNext(new GreetingRequest("joe-3")))
+                .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-3"))
+                .then(() -> requests.onComplete())
+                .expectComplete()
+                .verify(Duration.ofSeconds(3));
+    }
+
   private GreetingService createProxy(Microservices gateway) {
     return gateway.call().api(GreetingService.class); // create proxy for GreetingService API
   }
diff --git a/services/src/test/java/io/scalecube/services/ServiceRemoteTest.java b/services/src/test/java/io/scalecube/services/ServiceRemoteTest.java
@@ -27,10 +27,7 @@
 import org.junit.jupiter.api.Disabled;
 import org.junit.jupiter.api.Test;
 import org.reactivestreams.Publisher;
-import reactor.core.publisher.EmitterProcessor;
-import reactor.core.publisher.Flux;
-import reactor.core.publisher.Hooks;
-import reactor.core.publisher.Mono;
+import reactor.core.publisher.*;
 import reactor.test.StepVerifier;
 
 public class ServiceRemoteTest extends BaseTest {
@@ -369,6 +366,25 @@ public void test_remote_bidi_greeting_expect_IllegalArgumentException() {
         .verify(Duration.ofSeconds(3));
   }
 
+  @Test
+  public void test_remote_bidi_greeting_message_expect_IllegalArgumentException() {
+
+    // get a proxy to the service api.
+    GreetingService service = createProxy();
+
+    // call the service. bidiThrowingGreeting
+    Flux<GreetingResponse> responses =
+            service.bidiGreetingIllegalArgumentExceptionMessage(
+                    Mono.just(ServiceMessage.builder()
+                            .data(new GreetingRequest("IllegalArgumentException")).build()))
+            .map(ServiceMessage::data);
+
+    // call the service.
+    StepVerifier.create(responses)
+            .expectErrorMessage("IllegalArgumentException")
+            .verify(Duration.ofSeconds(3));
+  }
+
   @Test
   public void test_remote_bidi_greeting_expect_NotAuthorized() {
 
@@ -389,6 +405,25 @@ public void test_remote_bidi_greeting_expect_NotAuthorized() {
         .verify(Duration.ofSeconds(3));
   }
 
+  @Test
+  public void test_remote_bidi_greeting_message_expect_NotAuthorized() {
+
+    // get a proxy to the service api.
+    GreetingService service = createProxy();
+
+    DirectProcessor<GreetingRequest> requests = DirectProcessor.create();
+
+    // call the service.
+    Flux<GreetingResponse> responses = service.bidiGreetingNotAuthorizedMessage(
+            requests.map(request -> ServiceMessage.builder().data(request).build()))
+            .map(ServiceMessage::data);
+
+    StepVerifier.create(responses)
+            .then(() -> requests.onNext(new GreetingRequest("joe-1")))
+            .expectErrorMessage("Not authorized")
+            .verify(Duration.ofSeconds(3));
+  }
+
   @Test
   public void test_remote_bidi_greeting_expect_GreetingResponse() {
 
@@ -415,6 +450,31 @@ public void test_remote_bidi_greeting_expect_GreetingResponse() {
         .verify(Duration.ofSeconds(3));
   }
 
+  @Test
+  public void test_remote_bidi_greeting_message_expect_GreetingResponse() {
+
+    // get a proxy to the service api.
+    GreetingService service = createProxy();
+
+    UnicastProcessor<GreetingRequest> requests = UnicastProcessor.create();
+
+    // call the service.
+    Flux<GreetingResponse> responses = service.bidiGreetingMessage(requests
+            .map(request -> ServiceMessage.builder().data(request).build()))
+            .map(ServiceMessage::data);
+
+    StepVerifier.create(responses)
+            .then(() -> requests.onNext(new GreetingRequest("joe-1")))
+            .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-1"))
+            .then(() -> requests.onNext(new GreetingRequest("joe-2")))
+            .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-2"))
+            .then(() -> requests.onNext(new GreetingRequest("joe-3")))
+            .expectNextMatches(resp -> resp.getResult().equals(" hello to: joe-3"))
+            .then(() -> requests.onComplete())
+            .expectComplete()
+            .verify(Duration.ofSeconds(3));
+  }
+
   @Test
   public void test_services_contribute_to_cluster_metadata() {
     Map<String, String> tags = new HashMap<>();
diff --git a/services/src/test/java/io/scalecube/services/sut/GreetingService.java b/services/src/test/java/io/scalecube/services/sut/GreetingService.java
@@ -70,12 +70,27 @@ public interface GreetingService {
   @ServiceMethod
   Flux<GreetingResponse> bidiGreeting(Publisher<GreetingRequest> request);
 
+  @ServiceMethod
+  @RequestType(GreetingRequest.class)
+  @ResponseType(GreetingResponse.class)
+  Flux<ServiceMessage> bidiGreetingMessage(Publisher<ServiceMessage> requests);
+
   @ServiceMethod
   Flux<GreetingResponse> bidiGreetingNotAuthorized(Flux<GreetingRequest> request);
 
+  @ServiceMethod
+  @RequestType(GreetingRequest.class)
+  @ResponseType(GreetingResponse.class)
+  Flux<ServiceMessage> bidiGreetingNotAuthorizedMessage(Publisher<ServiceMessage> requests);
+
   @ServiceMethod
   Flux<GreetingResponse> bidiGreetingIllegalArgumentException(Publisher<GreetingRequest> request);
 
+  @ServiceMethod
+  @RequestType(GreetingRequest.class)
+  @ResponseType(GreetingResponse.class)
+  Flux<ServiceMessage> bidiGreetingIllegalArgumentExceptionMessage(Publisher<ServiceMessage> requests);
+
   @ServiceMethod
   Mono<GreetingResponse> greetingMonoEmpty(GreetingRequest request);
 
diff --git a/services/src/test/java/io/scalecube/services/sut/GreetingServiceImpl.java b/services/src/test/java/io/scalecube/services/sut/GreetingServiceImpl.java
@@ -78,17 +78,36 @@ public Flux<GreetingResponse> bidiGreeting(Publisher<GreetingRequest> request) {
         .map(onNext -> new GreetingResponse(" hello to: " + onNext.getName(), "" + instanceId));
   }
 
+  @Override
+  public Flux<ServiceMessage> bidiGreetingMessage(Publisher<ServiceMessage> requests) {
+    return Flux.from(requests).map(request -> {
+      GreetingRequest data = request.data();
+      GreetingResponse resp = new GreetingResponse(" hello to: " + data.getName(), "1");
+      return ServiceMessage.builder().data(resp).build();
+    });
+  }
+
   @Override
   public Flux<GreetingResponse> bidiGreetingNotAuthorized(Flux<GreetingRequest> request) {
     return Flux.error(new ForbiddenException("Not authorized"));
   }
 
+  @Override
+  public Flux<ServiceMessage> bidiGreetingNotAuthorizedMessage(Publisher<ServiceMessage> requests) {
+    return Flux.error(new ForbiddenException("Not authorized"));
+  }
+
   @Override
   public Flux<GreetingResponse> bidiGreetingIllegalArgumentException(
       Publisher<GreetingRequest> request) {
     throw new IllegalArgumentException("IllegalArgumentException");
   }
 
+  @Override
+  public Flux<ServiceMessage> bidiGreetingIllegalArgumentExceptionMessage(Publisher<ServiceMessage> requests) {
+    throw new IllegalArgumentException("IllegalArgumentException");
+  }
+
   @Override
   public Mono<GreetingResponse> greetingMonoEmpty(GreetingRequest request) {
     return Mono.empty();
