Support of file transfer (service to client) (#874)

Author: artem-v

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

@@ -12,7 +12,6 @@
 import io.scalecube.services.annotations.Service;
 import io.scalecube.services.annotations.ServiceMethod;
 import io.scalecube.services.annotations.Tag;
-import io.scalecube.services.api.Qualifier;
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.auth.Secured;
 import io.scalecube.services.methods.MethodInfo;
@@ -284,17 +283,6 @@ public static String restMethod(Method method) {
     return methodAnnotation != null ? methodAnnotation.value() : null;
   }
 
-  /**
-   * Handy method to get qualifier String from service's interface and method.
-   *
-   * @param serviceInterface service interface to get qualifier for
-   * @param method service's method to get qualifier for
-   * @return qualifier string
-   */
-  public static String qualifier(Class<?> serviceInterface, Method method) {
-    return Qualifier.asString(Reflect.serviceName(serviceInterface), Reflect.methodName(method));
-  }
-
   /**
    * Util function to perform basic validation of service message request.
    *
@@ -392,7 +380,11 @@ public static boolean isSecured(Method method) {
   }
 
   public static Scheduler executeOnScheduler(Method method, Map<String, Scheduler> schedulers) {
-    final Class<?> declaringClass = method.getDeclaringClass();
+    if (schedulers == null) {
+      return Schedulers.immediate();
+    }
+
+    final var declaringClass = method.getDeclaringClass();
 
     if (method.isAnnotationPresent(ExecuteOn.class)) {
       final var executeOn = method.getAnnotation(ExecuteOn.class);

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

@@ -13,8 +13,6 @@
 import io.scalecube.services.transport.api.ClientTransport;
 import java.lang.System.Logger;
 import java.lang.System.Logger.Level;
-import java.lang.reflect.InvocationHandler;
-import java.lang.reflect.Method;
 import java.lang.reflect.Proxy;
 import java.lang.reflect.Type;
 import java.util.Collections;
@@ -192,7 +190,7 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType
             () -> {
               ServiceMethodInvoker methodInvoker;
               if (serviceRegistry != null
-                  && (methodInvoker = serviceRegistry.getInvoker(request)) != null) {
+                  && (methodInvoker = serviceRegistry.lookupInvoker(request)) != null) {
                 // local service
                 return methodInvoker.invokeOne(request).map(this::throwIfError);
               } else {
@@ -246,7 +244,7 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseTyp
             () -> {
               ServiceMethodInvoker methodInvoker;
               if (serviceRegistry != null
-                  && (methodInvoker = serviceRegistry.getInvoker(request)) != null) {
+                  && (methodInvoker = serviceRegistry.lookupInvoker(request)) != null) {
                 // local service
                 return methodInvoker.invokeMany(request).map(this::throwIfError);
               } else {
@@ -301,7 +299,7 @@ public Flux<ServiceMessage> requestBidirectional(
                 ServiceMessage request = first.get();
                 ServiceMethodInvoker methodInvoker;
                 if (serviceRegistry != null
-                    && (methodInvoker = serviceRegistry.getInvoker(request)) != null) {
+                    && (methodInvoker = serviceRegistry.lookupInvoker(request)) != null) {
                   // local service
                   return methodInvoker.invokeBidirectional(messages).map(this::throwIfError);
                 } else {
@@ -328,59 +326,53 @@ public Flux<ServiceMessage> requestBidirectional(
    */
   @SuppressWarnings("unchecked")
   public <T> T api(Class<T> serviceInterface) {
-
-    final ServiceCall serviceCall = this;
-    final Map<Method, MethodInfo> genericReturnTypes = Reflect.methodsInfo(serviceInterface);
-
-    // noinspection unchecked,Convert2Lambda
     return (T)
         Proxy.newProxyInstance(
             getClass().getClassLoader(),
             new Class[] {serviceInterface},
-            new InvocationHandler() {
-              @Override
-              public Object invoke(Object proxy, Method method, Object[] params) {
-                Optional<Object> check =
-                    toStringOrEqualsOrHashCode(method.getName(), serviceInterface, params);
-                if (check.isPresent()) {
-                  return check.get(); // toString, hashCode was invoked.
-                }
+            (proxy, method, params) -> {
+              Optional<Object> check =
+                  toStringOrEqualsOrHashCode(method.getName(), serviceInterface, params);
+              if (check.isPresent()) {
+                return check.get(); // toString, hashCode was invoked.
+              }
 
-                final MethodInfo methodInfo = genericReturnTypes.get(method);
-                final Type returnType = methodInfo.parameterizedReturnType();
-                final boolean isServiceMessage = methodInfo.isReturnTypeServiceMessage();
-
-                Object request = methodInfo.requestType() == Void.TYPE ? null : params[0];
-
-                switch (methodInfo.communicationMode()) {
-                  case FIRE_AND_FORGET:
-                    return serviceCall.oneWay(toServiceMessage(methodInfo, request));
-
-                  case REQUEST_RESPONSE:
-                    return serviceCall
-                        .requestOne(toServiceMessage(methodInfo, request), returnType)
-                        .transform(asMono(isServiceMessage));
-
-                  case REQUEST_STREAM:
-                    return serviceCall
-                        .requestMany(toServiceMessage(methodInfo, request), returnType)
-                        .transform(asFlux(isServiceMessage));
-
-                  case REQUEST_CHANNEL:
-                    // this is REQUEST_CHANNEL so it means params[0] must
-                    // be a publisher - its safe to cast.
-                    //noinspection rawtypes
-                    return serviceCall
-                        .requestBidirectional(
-                            Flux.from((Publisher) request)
-                                .map(data -> toServiceMessage(methodInfo, data)),
-                            returnType)
-                        .transform(asFlux(isServiceMessage));
-
-                  default:
-                    throw new IllegalArgumentException(
-                        "Communication mode is not supported: " + method);
-                }
+              final var serviceCall = ServiceCall.this;
+              final var genericReturnTypes = Reflect.methodsInfo(serviceInterface);
+              final var methodInfo = genericReturnTypes.get(method);
+              final var returnType = methodInfo.parameterizedReturnType();
+              final var isServiceMessage = methodInfo.isReturnTypeServiceMessage();
+              final var request = methodInfo.requestType() == Void.TYPE ? null : params[0];
+
+              //noinspection EnhancedSwitchMigration
+              switch (methodInfo.communicationMode()) {
+                case FIRE_AND_FORGET:
+                  return serviceCall.oneWay(toServiceMessage(methodInfo, request));
+
+                case REQUEST_RESPONSE:
+                  return serviceCall
+                      .requestOne(toServiceMessage(methodInfo, request), returnType)
+                      .transform(asMono(isServiceMessage));
+
+                case REQUEST_STREAM:
+                  return serviceCall
+                      .requestMany(toServiceMessage(methodInfo, request), returnType)
+                      .transform(asFlux(isServiceMessage));
+
+                case REQUEST_CHANNEL:
+                  // this is REQUEST_CHANNEL so it means params[0] must
+                  // be a publisher - its safe to cast.
+                  //noinspection rawtypes
+                  return serviceCall
+                      .requestBidirectional(
+                          Flux.from((Publisher) request)
+                              .map(data -> toServiceMessage(methodInfo, data)),
+                          returnType)
+                      .transform(asFlux(isServiceMessage));
+
+                default:
+                  throw new IllegalArgumentException(
+                      "Communication mode is not supported: " + method);
               }
             });
   }

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

@@ -1,5 +1,7 @@
 package io.scalecube.services;
 
+import static io.scalecube.services.api.DynamicQualifier.isDynamicQualifier;
+
 import io.scalecube.services.api.DynamicQualifier;
 import io.scalecube.services.api.Qualifier;
 import java.util.Collections;
@@ -40,7 +42,7 @@ public ServiceReference(
     this.namespace = serviceRegistration.namespace();
     this.action = serviceMethodDefinition.action();
     this.qualifier = Qualifier.asString(namespace, action);
-    this.dynamicQualifier = qualifier.contains(":") ? new DynamicQualifier(qualifier) : null;
+    this.dynamicQualifier = isDynamicQualifier(qualifier) ? DynamicQualifier.from(qualifier) : null;
     this.contentTypes = Collections.unmodifiableSet(serviceEndpoint.contentTypes());
     this.tags = mergeTags(serviceMethodDefinition, serviceRegistration, serviceEndpoint);
     this.address = serviceEndpoint.address();

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

@@ -9,69 +9,125 @@
 import java.util.StringJoiner;
 import java.util.regex.Pattern;
 
+/**
+ * Representation of dynamic qualifier. Being used in service method definitions along with normal
+ * qualifiers. Typical example of dynamic qualifiers:
+ *
+ * <ul>
+ *   <li>v1/api/users/:userId
+ *   <li>v1/api/orders/:orderId/
+ *   <li>v1/api/categories/:categoryId/products/:productId
+ * </ul>
+ */
 public final class DynamicQualifier {
 
+  private static final Pattern DYNAMIC_QUALIFIER_PATTERN = Pattern.compile("(^|/):\\w+(?:/|$)");
+
   private final String qualifier;
   private final Pattern pattern;
   private final List<String> pathVariables;
   private final int size;
 
-  public DynamicQualifier(String qualifier) {
-    if (!qualifier.contains(":")) {
-      throw new IllegalArgumentException("Illegal dynamic qualifier: " + qualifier);
-    }
+  private DynamicQualifier(String qualifier) {
+    final var list = new ArrayList<String>();
+    final var builder = new StringBuilder();
 
-    final var pathVariables = new ArrayList<String>();
-    final var sb = new StringBuilder();
     for (var s : qualifier.split("/")) {
       if (s.startsWith(":")) {
         final var pathVar = s.substring(1);
-        sb.append("(?<").append(pathVar).append(">.*?)");
-        pathVariables.add(pathVar);
+        builder.append("(?<").append(pathVar).append(">.+)");
+        list.add(pathVar);
       } else {
-        sb.append(s);
+        builder.append(s);
       }
-      sb.append("/");
+      builder.append("/");
     }
-    sb.setLength(sb.length() - 1);
+    builder.setLength(builder.length() - 1);
 
     this.qualifier = qualifier;
-    this.pattern = Pattern.compile(sb.toString());
-    this.pathVariables = Collections.unmodifiableList(pathVariables);
+    this.pattern = Pattern.compile(builder.toString());
+    this.pathVariables = Collections.unmodifiableList(list);
     this.size = sizeOf(qualifier);
   }
 
+  /**
+   * Creates new {@link DynamicQualifier} instance.
+   *
+   * @param qualifier qualifier
+   * @return {@link DynamicQualifier} instance
+   */
+  public static DynamicQualifier from(String qualifier) {
+    return new DynamicQualifier(qualifier);
+  }
+
+  /**
+   * Returns whether given qualifier is dynamic qualifier or not.
+   *
+   * @param qualifier qualifier
+   * @return result
+   */
+  public static boolean isDynamicQualifier(String qualifier) {
+    return DYNAMIC_QUALIFIER_PATTERN.matcher(qualifier).find();
+  }
+
+  /**
+   * Original qualifier.
+   *
+   * @return result
+   */
   public String qualifier() {
     return qualifier;
   }
 
+  /**
+   * Compiled pattern.
+   *
+   * @return result
+   */
   public Pattern pattern() {
     return pattern;
   }
 
+  /**
+   * Returns path variable names.
+   *
+   * @return path variable names
+   */
   public List<String> pathVariables() {
     return pathVariables;
   }
 
+  /**
+   * Size of qualifier. This is a number of {@code /} symbols.
+   *
+   * @return result
+   */
   public int size() {
     return size;
   }
 
-  public Map<String, String> matchQualifier(String input) {
-    if (size != sizeOf(input)) {
+  /**
+   * Matches input qualifier against this dynamic qualifier.
+   *
+   * @param qualifier qualifier
+   * @return matched path variables key-value map, or null if no matching occurred
+   */
+  public Map<String, String> matchQualifier(String qualifier) {
+    if (size != sizeOf(qualifier)) {
       return null;
     }
 
-    final var matcher = pattern.matcher(input);
+    final var matcher = pattern.matcher(qualifier);
     if (!matcher.matches()) {
       return null;
     }
 
     final var map = new LinkedHashMap<String, String>();
     for (var pathVar : pathVariables) {
       final var value = matcher.group(pathVar);
-      Objects.requireNonNull(
-          value, "Path variable value must not be null, path variable: " + pathVar);
+      if (value == null || value.isEmpty()) {
+        throw new IllegalArgumentException("Wrong path variable: " + pathVar);
+      }
       map.put(pathVar, value);
     }