Merge pull request #737 from scalecube/update-authenticator-interface

Update authenticator interface

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

@@ -15,8 +15,8 @@ public class ServiceInfo {
   private final Map<String, String> tags;
   private final ServiceProviderErrorMapper errorMapper;
   private final ServiceMessageDataDecoder dataDecoder;
-  private final Authenticator authenticator;
-  private final PrincipalMapper<Object> principalMapper;
+  private final Authenticator<Object> authenticator;
+  private final PrincipalMapper<Object, Object> principalMapper;
 
   private ServiceInfo(Builder builder) {
     this.serviceInstance = builder.serviceInstance;
@@ -51,11 +51,11 @@ public ServiceMessageDataDecoder dataDecoder() {
     return dataDecoder;
   }
 
-  public Authenticator authenticator() {
+  public Authenticator<Object> authenticator() {
     return authenticator;
   }
 
-  public PrincipalMapper<Object> principalMapper() {
+  public PrincipalMapper<Object, Object> principalMapper() {
     return principalMapper;
   }
 
@@ -77,8 +77,8 @@ public static class Builder {
     private Map<String, String> tags = new HashMap<>();
     private ServiceProviderErrorMapper errorMapper;
     private ServiceMessageDataDecoder dataDecoder;
-    private Authenticator authenticator;
-    private PrincipalMapper<Object> principalMapper;
+    private Authenticator<Object> authenticator;
+    private PrincipalMapper<Object, Object> principalMapper;
 
     private Builder(ServiceInfo serviceInfo) {
       this.serviceInstance = serviceInfo.serviceInstance;
@@ -108,14 +108,16 @@ public Builder dataDecoder(ServiceMessageDataDecoder dataDecoder) {
       return this;
     }
 
-    public Builder authenticator(Authenticator authenticator) {
-      this.authenticator = authenticator;
+    @SuppressWarnings("unchecked")
+    public <T> Builder authenticator(Authenticator<? extends T> authenticator) {
+      this.authenticator = (Authenticator<Object>) authenticator;
       return this;
     }
 
     @SuppressWarnings("unchecked")
-    public <T> Builder principalMapper(PrincipalMapper<? extends T> principalMapper) {
-      this.principalMapper = (PrincipalMapper<Object>) principalMapper;
+    public <A, T> Builder principalMapper(
+        PrincipalMapper<? extends A, ? extends T> principalMapper) {
+      this.principalMapper = (PrincipalMapper<Object, Object>) principalMapper;
       return this;
     }
 
@@ -133,14 +135,14 @@ Builder dataDecoderIfAbsent(ServiceMessageDataDecoder dataDecoder) {
       return this;
     }
 
-    Builder authenticatorIfAbsent(Authenticator authenticator) {
+    Builder authenticatorIfAbsent(Authenticator<Object> authenticator) {
       if (this.authenticator == null) {
         this.authenticator = authenticator;
       }
       return this;
     }
 
-    Builder principalMapperIfAbsent(PrincipalMapper<Object> principalMapper) {
+    Builder principalMapperIfAbsent(PrincipalMapper<Object, Object> principalMapper) {
       if (this.principalMapper == null) {
         this.principalMapper = principalMapper;
       }

services-api/src/main/java/io/scalecube/services/auth/Authenticator.java

@@ -4,13 +4,19 @@
 import reactor.core.publisher.Mono;
 
 @FunctionalInterface
-public interface Authenticator {
+public interface Authenticator<A> {
+
+  /**
+   * Key in {@link reactor.util.context.Context} to represent authentication result after call to
+   * {@link #authenticate(Map)}.
+   */
+  String AUTH_CONTEXT_KEY = "auth.context";
 
   /**
    * Returns {@code authData} by given credentials.
    *
    * @param credentials credentials
    * @return async result with obtained {@code authData}
    */
-  Mono<Map<String, String>> authenticate(Map<String, String> credentials);
+  Mono<A> authenticate(Map<String, String> credentials);
 }

services-api/src/main/java/io/scalecube/services/auth/PrincipalMapper.java

@@ -1,15 +1,13 @@
 package io.scalecube.services.auth;
 
-import java.util.Map;
-
 @FunctionalInterface
-public interface PrincipalMapper<O> {
+public interface PrincipalMapper<A, P> {
 
   /**
    * Turns {@code authData} to concrete principal object.
    *
    * @param authData auth data
    * @return converted principle object
    */
-  O map(Map<String, String> authData);
+  P map(A authData);
 }

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

@@ -11,7 +11,6 @@
 import java.lang.reflect.InvocationTargetException;
 import java.lang.reflect.Method;
 import java.util.Collections;
-import java.util.Map;
 import java.util.Objects;
 import java.util.Optional;
 import java.util.StringJoiner;
@@ -31,8 +30,8 @@ public final class ServiceMethodInvoker {
   private final MethodInfo methodInfo;
   private final ServiceProviderErrorMapper errorMapper;
   private final ServiceMessageDataDecoder dataDecoder;
-  private final Authenticator authenticator;
-  private final PrincipalMapper<Object> principalMapper;
+  private final Authenticator<Object> authenticator;
+  private final PrincipalMapper<Object, Object> principalMapper;
 
   /**
    * Constructs a service method invoker out of real service object instance and method info.
@@ -51,8 +50,8 @@ public ServiceMethodInvoker(
       MethodInfo methodInfo,
       ServiceProviderErrorMapper errorMapper,
       ServiceMessageDataDecoder dataDecoder,
-      Authenticator authenticator,
-      PrincipalMapper<Object> principalMapper) {
+      Authenticator<Object> authenticator,
+      PrincipalMapper<Object, Object> principalMapper) {
     this.method = Objects.requireNonNull(method, "method");
     this.service = Objects.requireNonNull(service, "service");
     this.methodInfo = Objects.requireNonNull(methodInfo, "methodInfo");
@@ -69,7 +68,7 @@ public ServiceMethodInvoker(
    * @return mono of service message
    */
   public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
-    return Mono.defer(() -> authenticate(message))
+    return Mono.deferWithContext(context -> authenticate(message, context))
         .flatMap(authData -> deferWithContextOne(message, authData))
         .map(response -> toResponse(response, message.dataFormat()))
         .onErrorResume(throwable -> Mono.just(errorMapper.toMessage(throwable)));
@@ -82,7 +81,7 @@ public Mono<ServiceMessage> invokeOne(ServiceMessage message) {
    * @return flux of service messages
    */
   public Flux<ServiceMessage> invokeMany(ServiceMessage message) {
-    return Mono.defer(() -> authenticate(message))
+    return Mono.deferWithContext(context -> authenticate(message, context))
         .flatMapMany(authData -> deferWithContextMany(message, authData))
         .map(response -> toResponse(response, message.dataFormat()))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
@@ -98,24 +97,23 @@ public Flux<ServiceMessage> invokeBidirectional(Publisher<ServiceMessage> publis
     return Flux.from(publisher)
         .switchOnFirst(
             (first, messages) ->
-                Mono.defer(() -> authenticate(first.get()))
+                Mono.deferWithContext(context -> authenticate(first.get(), context))
                     .flatMapMany(authData -> deferWithContextBidirectional(messages, authData))
                     .map(response -> toResponse(response, first.get().dataFormat())))
         .onErrorResume(throwable -> Flux.just(errorMapper.toMessage(throwable)));
   }
 
-  private Mono<?> deferWithContextOne(ServiceMessage message, Map<String, String> authData) {
+  private Mono<?> deferWithContextOne(ServiceMessage message, Object authData) {
     return Mono.deferWithContext(context -> Mono.from(invoke(toRequest(message))))
         .subscriberContext(context -> newPrincipalContext(authData, context));
   }
 
-  private Flux<?> deferWithContextMany(ServiceMessage message, Map<String, String> authData) {
+  private Flux<?> deferWithContextMany(ServiceMessage message, Object authData) {
     return Flux.deferWithContext(context -> Flux.from(invoke(toRequest(message))))
         .subscriberContext(context -> newPrincipalContext(authData, context));
   }
 
-  private Flux<?> deferWithContextBidirectional(
-      Flux<ServiceMessage> messages, Map<String, String> authData) {
+  private Flux<?> deferWithContextBidirectional(Flux<ServiceMessage> messages, Object authData) {
     return Flux.deferWithContext(context -> messages.map(this::toRequest).transform(this::invoke))
         .subscriberContext(context -> newPrincipalContext(authData, context));
   }
@@ -149,10 +147,13 @@ private Object[] prepareArguments(Object request) {
     return arguments;
   }
 
-  private Mono<Map<String, String>> authenticate(ServiceMessage message) {
+  private Mono<Object> authenticate(ServiceMessage message, Context context) {
     if (!methodInfo.isAuth()) {
       return Mono.just(Collections.emptyMap());
     }
+    if (context.hasKey(Authenticator.AUTH_CONTEXT_KEY)) {
+      return Mono.just(context.get(Authenticator.AUTH_CONTEXT_KEY));
+    }
     if (authenticator == null) {
       throw new UnauthorizedException("Authenticator not found");
     }
@@ -201,12 +202,12 @@ private ServiceMessage toResponse(Object response, String dataFormat) {
         .build();
   }
 
-  private Context newPrincipalContext(Map<String, String> authData, Context context) {
+  private Context newPrincipalContext(Object authData, Context context) {
     if (principalMapper == null) {
       return context;
     }
     Object value = principalMapper.map(authData);
-    return Context.of(value.getClass(), value);
+    return Context.of(Authenticator.AUTH_CONTEXT_KEY, value);
   }
 
   public Object service() {