Simplify AmqpAdapterSaslAuthenticatorFactory.

Store collaborators in factory class' fields only.

Signed-off-by: Kai Hudalla <[email protected]>

Author: Kai Hudalla

adapters/amqp-vertx/src/main/java/org/eclipse/hono/adapter/amqp/AmqpAdapterSaslAuthenticatorFactory.java

@@ -20,6 +20,7 @@
 import java.util.HashMap;
 import java.util.Map;
 import java.util.Objects;
+import java.util.function.Function;
 import java.util.function.Supplier;
 
 import javax.net.ssl.SSLPeerUnverifiedException;
@@ -41,7 +42,6 @@
 import org.eclipse.hono.service.auth.device.UsernamePasswordCredentials;
 import org.eclipse.hono.service.auth.device.X509AuthProvider;
 import org.eclipse.hono.service.limiting.ConnectionLimitManager;
-import org.eclipse.hono.service.plan.ResourceLimitChecks;
 import org.eclipse.hono.tracing.TracingHelper;
 import org.eclipse.hono.util.AuthenticationConstants;
 import org.eclipse.hono.util.Constants;
@@ -69,20 +69,21 @@
  * <p>
  * On successful authentication of the device, a {@link Device} reflecting the device's credentials (as obtained from
  * the Credentials service) is stored in the attachments record of the {@code ProtonConnection} under key
- * {@link AmqpAdapterConstants#KEY_CLIENT_DEVICE}. The credentials supplied by the client is verified against the credentials that
- * the credentials service has on record for the device.
+ * {@link AmqpAdapterConstants#KEY_CLIENT_DEVICE}. The credentials supplied by the client are verified against
+ * the credentials that the Credentials service has on record for the device.
  * <p>
  * This factory supports the SASL PLAIN and SASL EXTERNAL mechanisms for authenticating devices.
  */
 public class AmqpAdapterSaslAuthenticatorFactory implements ProtonSaslAuthenticatorFactory {
 
     private final ProtocolAdapterProperties config;
     private final TenantClientFactory tenantClientFactory;
-    private final CredentialsClientFactory credentialsClientFactory;
-    private final Tracer tracer;
     private final Supplier<Span> spanFactory;
-    private final ConnectionLimitManager connectionLimitManager;
-    private final ResourceLimitChecks resourceLimitChecks;
+    private final ConnectionLimitManager adapterConnectionLimit;
+    private final Function<TenantObject, Future<Void>> tenantConnectionLimit;
+    private final DeviceCertificateValidator certValidator;
+    private final HonoClientBasedAuthProvider<UsernamePasswordCredentials> usernamePasswordAuthProvider;
+    private final HonoClientBasedAuthProvider<SubjectDnCredentials> clientCertAuthProvider;
 
     /**
      * Creates a new SASL authenticator factory for an authentication provider. If the AMQP adapter supports
@@ -95,10 +96,9 @@ public class AmqpAdapterSaslAuthenticatorFactory implements ProtonSaslAuthentica
      * @param tracer The tracer instance.
      * @param spanFactory The factory to use for creating and starting an OpenTracing span to
      *                    trace the authentication of the device.
-     * @param connectionLimitManager The connection limit manager to use to monitor the number of connections.
-     * @param resourceLimitChecks The resource limit checks instance to check if the maximum number of connections are
-     *            exceeded or not.
-     *
+     * @param adapterConnectionLimit The adapter level connection limit to enforce.
+     * @param tenantConnectionLimit The tenant level connection limit to enforce. The function must return
+     *                              a succeeded future if the connection limit has not been reached yet.
      * @throws NullPointerException if any of the parameters are null.
      */
     public AmqpAdapterSaslAuthenticatorFactory(
@@ -107,69 +107,43 @@ public AmqpAdapterSaslAuthenticatorFactory(
             final ProtocolAdapterProperties config,
             final Tracer tracer,
             final Supplier<Span> spanFactory,
-            final ConnectionLimitManager connectionLimitManager,
-            final ResourceLimitChecks resourceLimitChecks) {
+            final ConnectionLimitManager adapterConnectionLimit,
+            final Function<TenantObject, Future<Void>> tenantConnectionLimit) {
+
 
+        Objects.requireNonNull(credentialsClientFactory, "Credentials client cannot be null");
+        Objects.requireNonNull(tracer);
         this.tenantClientFactory = Objects.requireNonNull(tenantClientFactory, "Tenant client factory cannot be null");
-        this.credentialsClientFactory = Objects.requireNonNull(credentialsClientFactory, "Credentials client factory cannot be null");
         this.config = Objects.requireNonNull(config, "configuration cannot be null");
-        this.tracer = Objects.requireNonNull(tracer);
         this.spanFactory = Objects.requireNonNull(spanFactory);
-        this.connectionLimitManager = Objects.requireNonNull(connectionLimitManager);
-        this.resourceLimitChecks = Objects.requireNonNull(resourceLimitChecks);
+        this.adapterConnectionLimit = Objects.requireNonNull(adapterConnectionLimit);
+        this.tenantConnectionLimit = Objects.requireNonNull(tenantConnectionLimit);
+        this.certValidator = new DeviceCertificateValidator();
+        this.clientCertAuthProvider = new X509AuthProvider(credentialsClientFactory, config, tracer);
+        this.usernamePasswordAuthProvider = new UsernamePasswordAuthProvider(credentialsClientFactory, config, tracer);
     }
 
     @Override
     public ProtonSaslAuthenticator create() {
-        return new AmqpAdapterSaslAuthenticator(
-                tenantClientFactory,
-                credentialsClientFactory,
-                config,
-                tracer,
-                spanFactory.get(),
-                connectionLimitManager,
-                resourceLimitChecks);
+        return new AmqpAdapterSaslAuthenticator(spanFactory.get());
     }
 
     /**
      * Manage the SASL authentication process for the AMQP adapter.
      */
-    static final class AmqpAdapterSaslAuthenticator implements ProtonSaslAuthenticator {
+    final class AmqpAdapterSaslAuthenticator implements ProtonSaslAuthenticator {
 
-        private static final Logger LOG = LoggerFactory.getLogger(AmqpAdapterSaslAuthenticator.class);
+        private final Logger LOG = LoggerFactory.getLogger(getClass());
 
-        private final ProtocolAdapterProperties config;
-        private final TenantClientFactory tenantClientFactory;
-        private final CredentialsClientFactory credentialsClientFactory;
-        private final Tracer tracer;
         private final Span currentSpan;
-        private final ConnectionLimitManager connectionLimitManager;
-        private final ResourceLimitChecks resourceLimitChecks;
 
         private Sasl sasl;
         private boolean succeeded;
         private ProtonConnection protonConnection;
         private Certificate[] peerCertificateChain;
-        private HonoClientBasedAuthProvider<UsernamePasswordCredentials> usernamePasswordAuthProvider;
-        private HonoClientBasedAuthProvider<SubjectDnCredentials> clientCertAuthProvider;
-        private DeviceCertificateValidator certValidator;
-
-        AmqpAdapterSaslAuthenticator(
-                final TenantClientFactory tenantClientFactory,
-                final CredentialsClientFactory credentialsClientFactory,
-                final ProtocolAdapterProperties config,
-                final Tracer tracer,
-                final Span currentSpan,
-                final ConnectionLimitManager connectionLimitManager,
-                final ResourceLimitChecks resourceLimitChecks) {
-
-            this.tenantClientFactory = tenantClientFactory;
-            this.credentialsClientFactory = credentialsClientFactory;
-            this.config = config;
-            this.tracer = tracer;
+
+        AmqpAdapterSaslAuthenticator(final Span currentSpan) {
             this.currentSpan = currentSpan;
-            this.connectionLimitManager = connectionLimitManager;
-            this.resourceLimitChecks = resourceLimitChecks;
         }
 
         @Override
@@ -200,7 +174,7 @@ public void process(final Handler<Boolean> completionHandler) {
                 return;
             }
 
-            if (connectionLimitManager.isLimitExceeded()) {
+            if (adapterConnectionLimit.isLimitExceeded()) {
                 LOG.debug("Connection limit exceeded, reject connection request");
                 sasl.done(SaslOutcome.PN_SASL_TEMP);
                 completionHandler.handle(Boolean.TRUE);
@@ -272,14 +246,13 @@ private void verifyPlain(final byte[] saslResponse, final Handler<AsyncResult<De
                     currentSpan.log(items);
 
                     final Future<DeviceUser> authenticationTracker = Future.future();
-                    getUsernamePasswordAuthProvider().authenticate(credentials, currentSpan.context(),
+                    usernamePasswordAuthProvider.authenticate(credentials, currentSpan.context(),
                             authenticationTracker);
                     authenticationTracker
                             .compose(user -> getTenantObject(credentials.getTenantId())
-                                    .compose(tenant -> CompositeFuture
-                                            .all(checkTenantIsEnabled(tenant),
-                                                    resourceLimitChecks.isConnectionLimitExceeded(tenant))
-                                            .map(ok -> user)))
+                                    .compose(tenant -> CompositeFuture.all(
+                                            checkTenantIsEnabled(tenant),
+                                            tenantConnectionLimit.apply(tenant)).map(ok -> user)))
                             .setHandler(completer);
                 }
             } catch (CredentialException e) {
@@ -309,7 +282,7 @@ private void verifyExternal(final Handler<AsyncResult<DeviceUser>> completer) {
                         .compose(tenant -> {
                             try {
                                 final TrustAnchor trustAnchor = tenantTracker.result().getTrustAnchor();
-                                return getValidator().validate(Collections.singletonList(deviceCert), trustAnchor);
+                                return certValidator.validate(Collections.singletonList(deviceCert), trustAnchor);
                             } catch(final GeneralSecurityException e) {
                                 LOG.debug("cannot retrieve trust anchor of tenant [{}]", tenant.getTenantId(), e);
                                 return Future.failedFuture(new CredentialException("validation of client certificate failed"));
@@ -319,13 +292,12 @@ private void verifyExternal(final Handler<AsyncResult<DeviceUser>> completer) {
                             final Future<DeviceUser> user = Future.future();
                             final String tenantId = tenantTracker.result().getTenantId();
                             final SubjectDnCredentials credentials = SubjectDnCredentials.create(tenantId, deviceCert.getSubjectX500Principal());
-                            getCertificateAuthProvider().authenticate(credentials, currentSpan.context(), user);
+                            clientCertAuthProvider.authenticate(credentials, currentSpan.context(), user);
                             return user;
                         })
-                        .compose(user -> CompositeFuture
-                                .all(checkTenantIsEnabled(tenantTracker.result()),
-                                        resourceLimitChecks.isConnectionLimitExceeded(tenantTracker.result()))
-                                .map(ok -> user))
+                        .compose(user -> CompositeFuture.all(
+                                checkTenantIsEnabled(tenantTracker.result()),
+                                tenantConnectionLimit.apply(tenantTracker.result())).map(ok -> user))
                         .setHandler(completer);
             }
         }
@@ -347,26 +319,5 @@ private Future<TenantObject> checkTenantIsEnabled(final TenantObject tenant) {
                 return Future.failedFuture(new CredentialException("AMQP adapter is disabled for tenant"));
             }
         }
-
-        private HonoClientBasedAuthProvider<UsernamePasswordCredentials> getUsernamePasswordAuthProvider() {
-            if (usernamePasswordAuthProvider == null) {
-                usernamePasswordAuthProvider = new UsernamePasswordAuthProvider(credentialsClientFactory, config, tracer);
-            }
-            return usernamePasswordAuthProvider;
-        }
-
-        private HonoClientBasedAuthProvider<SubjectDnCredentials> getCertificateAuthProvider() {
-            if (clientCertAuthProvider == null) {
-                clientCertAuthProvider = new X509AuthProvider(credentialsClientFactory, config, tracer);
-            }
-            return clientCertAuthProvider;
-        }
-
-        private DeviceCertificateValidator getValidator() {
-            if (certValidator == null) {
-                certValidator = new DeviceCertificateValidator();
-            }
-            return certValidator;
-        }
     }
 }

adapters/amqp-vertx/src/main/java/org/eclipse/hono/adapter/amqp/impl/VertxBasedAmqpProtocolAdapter.java

@@ -162,7 +162,7 @@ protected void doStart(final Future<Void> startFuture) {
                                     .withTag(Tags.COMPONENT.getKey(), getTypeName())
                                     .start(),
                                 connectionLimitManager,
-                                getResourceLimitChecks());
+                                this::checkConnectionLimit);
                     }
                     return Future.succeededFuture();
                 }).compose(succcess -> {

adapters/mqtt-vertx-base/src/main/java/org/eclipse/hono/adapter/mqtt/AbstractVertxBasedMqttProtocolAdapter.java

@@ -439,8 +439,9 @@ private Future<Device> handleEndpointConnectionWithAuthentication(final MqttEndp
         return authAttempt
                 .compose(authenticatedDevice -> CompositeFuture.all(
                         getTenantConfiguration(authenticatedDevice.getTenantId(), currentSpan.context())
-                                .compose(tenantObj -> CompositeFuture
-                                        .all(isAdapterEnabled(tenantObj), getResourceLimitChecks().isConnectionLimitExceeded(tenantObj))),
+                            .compose(tenantObj -> CompositeFuture.all(
+                                    isAdapterEnabled(tenantObj),
+                                    checkConnectionLimit(tenantObj))),
                         checkDeviceRegistration(authenticatedDevice, currentSpan.context()))
                         .map(ok -> authenticatedDevice))
                 .compose(authenticatedDevice -> createLinks(authenticatedDevice, currentSpan))

service-base/src/main/java/org/eclipse/hono/service/AbstractProtocolAdapterBase.java

@@ -485,6 +485,27 @@ protected final Future<TenantObject> isAdapterEnabled(final TenantObject tenantC
         }
     }
 
+    /**
+     * Checks if this adapter may accept another connection from a device.
+     * <p>
+     * This default implementation uses the
+     * {@link ResourceLimitChecks#isConnectionLimitExceeded(TenantObject)} method
+     * to verify if the tenant's connection limit has been reached.
+     * 
+     * @param tenantConfig The tenant to check the connection limit for.
+     * @return A succeeded future if the connection limit has not been reached yet
+     *         or if the limits could not be checked.
+     *         Otherwise the future will be failed with a {@link ClientErrorException}
+     *         containing the 403 Forbidden status code.
+     * @throws NullPointerException if tenant is {@code null}.
+     */
+    protected Future<Void> checkConnectionLimit(final TenantObject tenantConfig) {
+
+        Objects.requireNonNull(tenantConfig);
+        return resourceLimitChecks.isConnectionLimitExceeded(tenantConfig)
+                .map(r -> (Void) null);
+    }
+
     /**
      * Validates a message's target address for consistency with Hono's addressing rules.
      *