AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { final void handleEndpointConnection(final MqttEndpoint endpoint) { log.debug("connection request from client [client-id: {}]", endpoint.clientIdentifier()); final Span span = tracer.buildSpan("CONNECT") .ignoreActiveSpan() .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_SERVER) .withTag(Tags.COMPONENT.getKey(), getTypeName()) .withTag(TracingHelper.TAG_CLIENT_ID.getKey(), endpoint.clientIdentifier()) .start(); if (!endpoint.isCleanSession()) { span.log("ignoring client's intent to resume existing session"); } if (endpoint.will() != null) { span.log("ignoring client's last will"); } final AtomicBoolean connectionClosedPrematurely = new AtomicBoolean(false); endpoint.closeHandler(v -> { log.debug("client closed connection before CONNACK got sent to client [client-id: {}]", endpoint.clientIdentifier()); TracingHelper.logError(span, "client closed connection"); connectionClosedPrematurely.set(true); }); isConnected() .compose(v -> handleConnectionRequest(endpoint, span)) .compose(authenticatedDevice -> handleConnectionRequestResult(endpoint, authenticatedDevice, connectionClosedPrematurely, span)) .onSuccess(authenticatedDevice -> { endpoint.accept(false); span.log("connection accepted"); metrics.reportConnectionAttempt( ConnectionAttemptOutcome.SUCCEEDED, Optional.ofNullable(authenticatedDevice).map(device -> device.getTenantId()).orElse(null)); }) .onFailure(t -> { log.debug("rejecting connection request from client [clientId: {}], cause:", endpoint.clientIdentifier(), t); final MqttConnectReturnCode code = getConnectReturnCode(t); rejectConnectionRequest(endpoint, code, span); TracingHelper.logError(span, t); reportFailedConnectionAttempt(t); }) .onComplete(result -> span.finish()); }  final void setAuthHandler(final AuthHandler<MqttConnectContext> authHandler); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue(); @Autowired final void setMetrics(final MqttAdapterMetrics metrics); void setMqttSecureServer(final MqttServer server); void setMqttInsecureServer(final MqttServer server); final Future<Void> uploadMessage(
            final MqttContext ctx,
            final ResourceIdentifier resource,
            final MqttPublishMessage message); final Future<Void> uploadTelemetryMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadEventMessage(
            final MqttContext ctx,
            final String tenant,
            final String deviceId,
            final Buffer payload); final Future<Void> uploadCommandResponseMessage(
            final MqttContext ctx,
            final ResourceIdentifier targetAddress);  }
@SuppressWarnings("unchecked") @Test public void testConnectionMetrics() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser("DEFAULT_TENANT", "4711"))); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(metrics).incrementConnections("DEFAULT_TENANT"); final ArgumentCaptor<Handler<Void>> closeHandlerCaptor = ArgumentCaptor.forClass(Handler.class); verify(endpoint, times(2)).closeHandler(closeHandlerCaptor.capture()); closeHandlerCaptor.getValue().handle(null); verify(metrics).decrementConnections("DEFAULT_TENANT"); }
@SuppressWarnings("unchecked") @Test public void testUnauthenticatedConnectionMetrics() { config.setAuthenticationRequired(false); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); verify(authHandler, never()).authenticateDevice(any(MqttConnectContext.class)); verify(metrics).incrementUnauthenticatedConnections(); final ArgumentCaptor<Handler<Void>> closeHandlerCaptor = ArgumentCaptor.forClass(Handler.class); verify(endpoint, times(2)).closeHandler(closeHandlerCaptor.capture()); closeHandlerCaptor.getValue().handle(null); verify(metrics).decrementUnauthenticatedConnections(); }
@Test public void testConnectionsLimitExceeded() { final MqttServer server = getMqttServer(false); config.setAuthenticationRequired(true); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser(Constants.DEFAULT_TENANT, "4711"))); when(resourceLimitChecks.isConnectionLimitReached(any(TenantObject.class), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(authHandler).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.TENANT_CONNECTIONS_EXCEEDED, Constants.DEFAULT_TENANT); }
@Test public void testConnectionDurationLimitExceeded() { final MqttServer server = getMqttServer(false); config.setAuthenticationRequired(true); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser(Constants.DEFAULT_TENANT, "4711"))); when(resourceLimitChecks.isConnectionDurationLimitReached(any(TenantObject.class), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(authHandler).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.CONNECTION_DURATION_EXCEEDED, Constants.DEFAULT_TENANT); }
@Test public void testEndpointHandlerFailsWithoutDownstreamConnections() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); }
@Test public void testEndpointHandlerAcceptsUnauthenticatedDevices() { config.setAuthenticationRequired(false); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); verify(endpoint).accept(false); verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.SUCCEEDED, null); }
@Test public void testEndpointHandlerRejectsDeviceOfDisabledTenant() { final MqttServer server = getMqttServer(false); final TenantObject myTenantConfig = TenantObject.from("my-tenant", true); myTenantConfig.addAdapter(new Adapter(ADAPTER_TYPE).setEnabled(Boolean.FALSE)); when(tenantClient.get(eq("my-tenant"), (SpanContext) any())).thenReturn(Future.succeededFuture(myTenantConfig)); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser("my-tenant", "4711"))); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED); verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.ADAPTER_DISABLED, "my-tenant"); }
@Test public void testEndpointHandlerRejectsUnauthenticatedDevices() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); when(authHandler.authenticateDevice(any(MqttConnectContext.class))).thenReturn( Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_UNAUTHORIZED))); forceClientMocksToConnected(); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); adapter.handleEndpointConnection(endpoint); adapter.handleEndpointConnection(endpoint); verify(authHandler, times(3)).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint, times(3)).reject(MqttConnectReturnCode.CONNECTION_REFUSED_BAD_USER_NAME_OR_PASSWORD); }
@Test public void testEndpointHandlerTriesToAuthenticateDevice() { final MqttServer server = getMqttServer(false); config.setAuthenticationRequired(true); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))).thenReturn( Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_UNAUTHORIZED))); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(authHandler).authenticateDevice(any(MqttConnectContext.class)); }
@Test public void testEndpointHandlerRejectsConnectionsExceedingAdapterLimit(final VertxTestContext ctx) { final MqttServer server = getMqttServer(false); config.setMaxConnections(1); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final Promise<Void> startupTracker = Promise.promise(); adapter.start(startupTracker); startupTracker.future().onComplete(ctx.succeeding(s -> { forceClientMocksToConnected(); when(metrics.getNumberOfConnections()).thenReturn(1); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); ctx.verify(() -> { verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_SERVER_UNAVAILABLE); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.ADAPTER_CONNECTIONS_EXCEEDED, null); }); ctx.completeNow(); })); }
@SuppressWarnings({ "unchecked" }) @Test public void testAuthenticatedMqttAdapterCreatesMessageHandlersForAuthenticatedDevices() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser(Constants.DEFAULT_TENANT, "4711"))); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(authHandler).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint).accept(false); verify(endpoint).publishHandler(any(Handler.class)); verify(endpoint, times(2)).closeHandler(any(Handler.class)); verify(metrics).reportConnectionAttempt(ConnectionAttemptOutcome.SUCCEEDED, Constants.DEFAULT_TENANT); }
@Test public void testAuthenticatedMqttAdapterRejectsConnectionForNonExistingDevice() { final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); when(authHandler.authenticateDevice(any(MqttConnectContext.class))) .thenReturn(Future.succeededFuture(new DeviceUser(Constants.DEFAULT_TENANT, "9999"))); when(regClient.assertRegistration(eq("9999"), (String) any(), (SpanContext) any())) .thenReturn(Future.failedFuture(new ClientErrorException( HttpURLConnection.HTTP_NOT_FOUND, "device unknown or disabled"))); final MqttEndpoint endpoint = getMqttEndpointAuthenticated(); adapter.handleEndpointConnection(endpoint); verify(authHandler).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint).reject(MqttConnectReturnCode.CONNECTION_REFUSED_NOT_AUTHORIZED); verify(metrics).reportConnectionAttempt( ConnectionAttemptOutcome.REGISTRATION_ASSERTION_FAILURE, Constants.DEFAULT_TENANT); }
@SuppressWarnings("unchecked") @Test public void testUnauthenticatedMqttAdapterCreatesMessageHandlersForAllDevices() { config.setAuthenticationRequired(false); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); forceClientMocksToConnected(); final MqttEndpoint endpoint = mockEndpoint(); adapter.handleEndpointConnection(endpoint); verify(authHandler, never()).authenticateDevice(any(MqttConnectContext.class)); verify(endpoint).publishHandler(any(Handler.class)); verify(endpoint, times(2)).closeHandler(any(Handler.class)); verify(endpoint).accept(false); }