Merge pull request #781 from scalecube/add-credentialsSupplier-api

Add credentials supplier api

Author: artem-v

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

@@ -1,8 +1,5 @@
 package io.scalecube.services;
 
-import static java.util.Objects.requireNonNull;
-
-import io.scalecube.net.Address;
 import io.scalecube.services.api.ErrorData;
 import io.scalecube.services.api.ServiceMessage;
 import io.scalecube.services.exceptions.DefaultErrorMapper;
@@ -162,17 +159,6 @@ public Mono<Void> oneWay(ServiceMessage request) {
     return Mono.defer(() -> requestOne(request, Void.class).then());
   }
 
-  /**
-   * Issues fire-and-forget request.
-   *
-   * @param request request message to send.
-   * @param address of remote target service to invoke.
-   * @return mono publisher completing normally or with error.
-   */
-  public Mono<Void> oneWay(ServiceMessage request, Address address) {
-    return Mono.defer(() -> requestOne(request, Void.class, address).then());
-  }
-
   /**
    * Issues request-and-reply request.
    *
@@ -187,48 +173,31 @@ public Mono<ServiceMessage> requestOne(ServiceMessage request) {
    * Issues request-and-reply request.
    *
    * @param request request message to send.
-   * @param responseType type of response.
+   * @param responseType type of response (optional).
    * @return mono publisher completing with single response message or with error.
    */
   public Mono<ServiceMessage> requestOne(ServiceMessage request, Type responseType) {
     return Mono.defer(
         () -> {
-          Objects.requireNonNull(request.qualifier(), "qualifier");
-
           ServiceMethodInvoker methodInvoker;
           if (methodRegistry != null
               && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
             // local service
             return methodInvoker.invokeOne(request).map(this::throwIfError);
           } else {
             // remote service
-            return Mono.fromCallable(() -> addressLookup(request))
-                .flatMap(address -> requestOne(request, responseType, address));
+            Objects.requireNonNull(transport, "[requestOne] transport");
+            return Mono.fromCallable(() -> serviceLookup(request))
+                .flatMap(
+                    serviceReference ->
+                        transport
+                            .create(serviceReference)
+                            .requestResponse(request, responseType)
+                            .map(this::throwIfError));
           }
         });
   }
 
-  /**
-   * Given an address issues request-and-reply request to a remote address.
-   *
-   * @param request request message to send.
-   * @param responseType type of response.
-   * @param address of remote target service to invoke.
-   * @return mono publisher completing with single response message or with error.
-   */
-  public Mono<ServiceMessage> requestOne(
-      ServiceMessage request, Type responseType, Address address) {
-    return Mono.defer(
-        () -> {
-          requireNonNull(address, "requestOne address parameter is required and must not be null");
-          requireNonNull(transport, "transport is required and must not be null");
-          return transport
-              .create(address)
-              .requestResponse(request, responseType)
-              .map(this::throwIfError);
-        });
-  }
-
   /**
    * Issues request to service which returns stream of service messages back.
    *
@@ -243,49 +212,31 @@ public Flux<ServiceMessage> requestMany(ServiceMessage request) {
    * Issues request to service which returns stream of service messages back.
    *
    * @param request request with given headers.
-   * @param responseType type of responses.
+   * @param responseType type of responses (optional).
    * @return flux publisher of service responses.
    */
   public Flux<ServiceMessage> requestMany(ServiceMessage request, Type responseType) {
     return Flux.defer(
         () -> {
-          Objects.requireNonNull(request.qualifier(), "qualifier");
-
           ServiceMethodInvoker methodInvoker;
           if (methodRegistry != null
               && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
             // local service
             return methodInvoker.invokeMany(request).map(this::throwIfError);
           } else {
             // remote service
-            return Mono.fromCallable(() -> addressLookup(request))
-                .flatMapMany(address -> requestMany(request, responseType, address));
+            Objects.requireNonNull(transport, "[requestMany] transport");
+            return Mono.fromCallable(() -> serviceLookup(request))
+                .flatMapMany(
+                    serviceReference ->
+                        transport
+                            .create(serviceReference)
+                            .requestStream(request, responseType)
+                            .map(this::throwIfError));
           }
         });
   }
 
-  /**
-   * Given an address issues request to remote service which returns stream of service messages
-   * back.
-   *
-   * @param request request with given headers.
-   * @param responseType type of responses.
-   * @param address of remote target service to invoke.
-   * @return flux publisher of service responses.
-   */
-  public Flux<ServiceMessage> requestMany(
-      ServiceMessage request, Type responseType, Address address) {
-    return Flux.defer(
-        () -> {
-          requireNonNull(address, "requestMany address parameter is required and must not be null");
-          requireNonNull(transport, "transport is required and must not be null");
-          return transport
-              .create(address)
-              .requestStream(request, responseType)
-              .map(this::throwIfError);
-        });
-  }
-
   /**
    * Issues stream of service requests to service which returns stream of service messages back.
    *
@@ -300,7 +251,7 @@ public Flux<ServiceMessage> requestBidirectional(Publisher<ServiceMessage> publi
    * Issues stream of service requests to service which returns stream of service messages back.
    *
    * @param publisher of service requests.
-   * @param responseType type of responses.
+   * @param responseType type of responses (optional).
    * @return flux publisher of service responses.
    */
   public Flux<ServiceMessage> requestBidirectional(
@@ -310,47 +261,27 @@ public Flux<ServiceMessage> requestBidirectional(
             (first, messages) -> {
               if (first.hasValue()) {
                 ServiceMessage request = first.get();
-                Objects.requireNonNull(request.qualifier(), "qualifier");
-
                 ServiceMethodInvoker methodInvoker;
                 if (methodRegistry != null
                     && (methodInvoker = methodRegistry.getInvoker(request.qualifier())) != null) {
                   // local service
                   return methodInvoker.invokeBidirectional(messages).map(this::throwIfError);
                 } else {
                   // remote service
-                  return Mono.fromCallable(() -> addressLookup(request))
+                  Objects.requireNonNull(transport, "[requestBidirectional] transport");
+                  return Mono.fromCallable(() -> serviceLookup(request))
                       .flatMapMany(
-                          address -> requestBidirectional(messages, responseType, address));
+                          serviceReference ->
+                              transport
+                                  .create(serviceReference)
+                                  .requestChannel(messages, responseType)
+                                  .map(this::throwIfError));
                 }
               }
               return messages;
             });
   }
 
-  /**
-   * Given an address issues stream of service requests to service which returns stream of service
-   * messages back.
-   *
-   * @param publisher of service requests.
-   * @param responseType type of responses.
-   * @param address of remote target service to invoke.
-   * @return flux publisher of service responses.
-   */
-  public Flux<ServiceMessage> requestBidirectional(
-      Publisher<ServiceMessage> publisher, Type responseType, Address address) {
-    return Flux.defer(
-        () -> {
-          requireNonNull(
-              address, "requestBidirectional address parameter is required and must not be null");
-          requireNonNull(transport, "transport is required and must not be null");
-          return transport
-              .create(address)
-              .requestChannel(publisher, responseType)
-              .map(this::throwIfError);
-        });
-  }
-
   /**
    * Create proxy creates a java generic proxy instance by a given service interface.
    *
@@ -416,10 +347,9 @@ public Object invoke(Object proxy, Method method, Object[] params) {
             });
   }
 
-  private Address addressLookup(ServiceMessage request) {
+  private ServiceReference serviceLookup(ServiceMessage request) {
     return router
         .route(serviceRegistry, request)
-        .map(ServiceReference::address)
         .orElseThrow(() -> noReachableMemberException(request));
   }
 

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

@@ -29,10 +29,11 @@ public class ServiceEndpoint implements Externalizable {
   private Collection<ServiceRegistration> serviceRegistrations;
 
   /**
-   * Constructor for SerDe.
+   * Constructor for de/serialization purpose.
    *
    * @deprecated exposed only for de/serialization purpose.
    */
+  @Deprecated
   public ServiceEndpoint() {}
 
   private ServiceEndpoint(Builder builder) {
@@ -112,7 +113,7 @@ public void writeExternal(ObjectOutput out) throws IOException {
     out.writeInt(tags.size());
     for (Entry<String, String> entry : tags.entrySet()) {
       out.writeUTF(entry.getKey());
-      out.writeUTF(entry.getValue());
+      out.writeObject(entry.getValue()); // value is nullable
     }
 
     // serviceRegistrations
@@ -143,7 +144,7 @@ public void readExternal(ObjectInput in) throws IOException, ClassNotFoundExcept
     Map<String, String> tags = new HashMap<>(tagsSize);
     for (int i = 0; i < tagsSize; i++) {
       String key = in.readUTF();
-      String value = in.readUTF();
+      String value = (String) in.readObject(); // value is nullable
       tags.put(key, value);
     }
     this.tags = Collections.unmodifiableMap(tags);

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

@@ -1,6 +1,5 @@
 package io.scalecube.services;
 
-import io.scalecube.services.auth.Authenticator;
 import io.scalecube.services.auth.PrincipalMapper;
 import io.scalecube.services.exceptions.ServiceProviderErrorMapper;
 import io.scalecube.services.transport.api.ServiceMessageDataDecoder;
@@ -16,15 +15,13 @@ public class ServiceInfo {
   private final Map<String, String> tags;
   private final ServiceProviderErrorMapper errorMapper;
   private final ServiceMessageDataDecoder dataDecoder;
-  private final Authenticator<Object> authenticator;
   private final PrincipalMapper<Object, Object> principalMapper;
 
   private ServiceInfo(Builder builder) {
     this.serviceInstance = builder.serviceInstance;
     this.tags = Collections.unmodifiableMap(new HashMap<>(builder.tags));
     this.errorMapper = builder.errorMapper;
     this.dataDecoder = builder.dataDecoder;
-    this.authenticator = builder.authenticator;
     this.principalMapper = builder.principalMapper;
   }
 
@@ -52,10 +49,6 @@ public ServiceMessageDataDecoder dataDecoder() {
     return dataDecoder;
   }
 
-  public Authenticator<Object> authenticator() {
-    return authenticator;
-  }
-
   public PrincipalMapper<Object, Object> principalMapper() {
     return principalMapper;
   }
@@ -67,26 +60,23 @@ public String toString() {
         .add("tags(" + tags.size() + ")")
         .add("errorMapper=" + errorMapper)
         .add("dataDecoder=" + dataDecoder)
-        .add("authenticator=" + authenticator)
         .add("principalMapper=" + principalMapper)
         .toString();
   }
 
   public static class Builder {
 
-    private Object serviceInstance;
-    private Map<String, String> tags = new HashMap<>();
+    private final Object serviceInstance;
+    private final Map<String, String> tags = new HashMap<>();
     private ServiceProviderErrorMapper errorMapper;
     private ServiceMessageDataDecoder dataDecoder;
-    private Authenticator<Object> authenticator;
     private PrincipalMapper<Object, Object> principalMapper;
 
     private Builder(ServiceInfo serviceInfo) {
       this.serviceInstance = serviceInfo.serviceInstance;
       this.tags.putAll(new HashMap<>(serviceInfo.tags));
       this.errorMapper = serviceInfo.errorMapper;
       this.dataDecoder = serviceInfo.dataDecoder;
-      this.authenticator = serviceInfo.authenticator;
       this.principalMapper = serviceInfo.principalMapper;
     }
 
@@ -131,31 +121,16 @@ public Builder dataDecoder(ServiceMessageDataDecoder dataDecoder) {
       return this;
     }
 
-    /**
-     * Setter for {@code authenticator}. Overrides default {@code Microservices.authenticator}.
-     *
-     * @param authenticator authenticator
-     * @param <T> type of auth data returned by authenticator
-     * @return this builder
-     */
-    @SuppressWarnings("unchecked")
-    public <T> Builder authenticator(Authenticator<? extends T> authenticator) {
-      Objects.requireNonNull(authenticator, "authenticator");
-      this.authenticator = (Authenticator<Object>) authenticator;
-      return this;
-    }
-
     /**
      * Setter for {@code principalMapper}. Overrides default {@code Microservices.principalMapper}.
      *
      * @param principalMapper principalMapper
-     * @param <A> type of auth data returned by authenticator
-     * @param <T> type of principal after mapping auth data to principal
+     * @param <T> auth data type
+     * @param <R> principal type
      * @return this builder
      */
     @SuppressWarnings("unchecked")
-    public <A, T> Builder principalMapper(
-        PrincipalMapper<? extends A, ? extends T> principalMapper) {
+    public <T, R> Builder principalMapper(PrincipalMapper<? super T, ? extends R> principalMapper) {
       Objects.requireNonNull(principalMapper, "principalMapper");
       this.principalMapper = (PrincipalMapper<Object, Object>) principalMapper;
       return this;
@@ -175,13 +150,6 @@ Builder dataDecoderIfAbsent(ServiceMessageDataDecoder dataDecoder) {
       return this;
     }
 
-    Builder authenticatorIfAbsent(Authenticator<Object> authenticator) {
-      if (this.authenticator == null) {
-        this.authenticator = authenticator;
-      }
-      return this;
-    }
-
     Builder principalMapperIfAbsent(PrincipalMapper<Object, Object> principalMapper) {
       if (this.principalMapper == null) {
         this.principalMapper = principalMapper;

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

@@ -24,10 +24,11 @@ public class ServiceMethodDefinition implements Externalizable {
   private boolean isSecured;
 
   /**
-   * Constructor for SerDe.
+   * Constructor for de/serialization purpose.
    *
    * @deprecated exposed only for de/serialization purpose.
    */
+  @Deprecated
   public ServiceMethodDefinition() {}
 
   /**

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

@@ -23,10 +23,11 @@ public class ServiceRegistration implements Externalizable {
   private Collection<ServiceMethodDefinition> methods;
 
   /**
-   * Constructor for SerDe.
+   * Constructor for de/serialization purpose.
    *
    * @deprecated exposed only for de/serialization purpose.
    */
+  @Deprecated
   public ServiceRegistration() {}
 
   /**