AbstractVertxBasedMqttProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final Future<Void> uploadCommandResponseMessage( final MqttContext ctx, final ResourceIdentifier targetAddress) { Objects.requireNonNull(ctx); Objects.requireNonNull(targetAddress); final String[] addressPath = targetAddress.getResourcePath(); Integer status = null; String reqId = null; final Future<CommandResponse> commandResponseTracker; if (addressPath.length <= CommandConstants.TOPIC_POSITION_RESPONSE_STATUS) { commandResponseTracker = Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "command response topic has too few segments")); } else { try { status = Integer.parseInt(addressPath[CommandConstants.TOPIC_POSITION_RESPONSE_STATUS]); } catch (final NumberFormatException e) { log.trace("got invalid status code [{}] [tenant-id: {}, device-id: {}]", addressPath[CommandConstants.TOPIC_POSITION_RESPONSE_STATUS], targetAddress.getTenantId(), targetAddress.getResourceId()); } if (status != null) { reqId = addressPath[CommandConstants.TOPIC_POSITION_RESPONSE_REQ_ID]; final CommandResponse commandResponse = CommandResponse.from(reqId, targetAddress.getTenantId(), targetAddress.getResourceId(), ctx.message().payload(), ctx.contentType(), status); commandResponseTracker = commandResponse != null ? Future.succeededFuture(commandResponse) : Future.failedFuture(new ClientErrorException( HttpURLConnection.HTTP_BAD_REQUEST, "command response topic contains invalid data")); } else { commandResponseTracker = Future.failedFuture( new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "invalid status code")); } } final Span currentSpan = TracingHelper .buildChildSpan(tracer, ctx.getTracingContext(), "upload Command response", getTypeName()) .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT) .withTag(TracingHelper.TAG_TENANT_ID, targetAddress.getTenantId()) .withTag(TracingHelper.TAG_DEVICE_ID, targetAddress.getResourceId()) .withTag(Constants.HEADER_COMMAND_RESPONSE_STATUS, status) .withTag(Constants.HEADER_COMMAND_REQUEST_ID, reqId) .withTag(TracingHelper.TAG_AUTHENTICATED.getKey(), ctx.authenticatedDevice() != null) .start(); final int payloadSize = Optional.ofNullable(ctx.message().payload()).map(Buffer::length).orElse(0); final Future<JsonObject> tokenTracker = getRegistrationAssertion(targetAddress.getTenantId(), targetAddress.getResourceId(), ctx.authenticatedDevice(), currentSpan.context()); final Future<TenantObject> tenantTracker = getTenantConfiguration(targetAddress.getTenantId(), ctx.getTracingContext()); final Future<TenantObject> tenantValidationTracker = CompositeFuture.all( isAdapterEnabled(tenantTracker.result()), checkMessageLimit(tenantTracker.result(), payloadSize, currentSpan.context())) .map(success -> tenantTracker.result()); return CompositeFuture.all(tenantTracker, commandResponseTracker) .compose(success -> CompositeFuture.all(tokenTracker, tenantValidationTracker)) .compose(ok -> sendCommandResponse(targetAddress.getTenantId(), commandResponseTracker.result(), currentSpan.context())) .compose(delivery -> { log.trace("successfully forwarded command response from device [tenant-id: {}, device-id: {}]", targetAddress.getTenantId(), targetAddress.getResourceId()); metrics.reportCommand( Direction.RESPONSE, targetAddress.getTenantId(), tenantTracker.result(), ProcessingOutcome.FORWARDED, payloadSize, ctx.getTimer()); if (ctx.deviceEndpoint().isConnected() && ctx.message().qosLevel() == MqttQoS.AT_LEAST_ONCE) { ctx.deviceEndpoint().publishAcknowledge(ctx.message().messageId()); } currentSpan.finish(); return Future.<Void> succeededFuture(); }) .recover(t -> { TracingHelper.logError(currentSpan, t); currentSpan.finish(); metrics.reportCommand( Direction.RESPONSE, targetAddress.getTenantId(), tenantTracker.result(), ProcessingOutcome.from(t), payloadSize, ctx.getTimer()); return Future.failedFuture(t); }); }  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);  }
@Test public void testUploadEmptyCommandResponseSucceeds(final VertxTestContext ctx) { final CommandResponseSender sender = mock(CommandResponseSender.class); when(sender.sendCommandResponse(any(CommandResponse.class), (SpanContext) any())) .thenReturn(Future.succeededFuture(mock(ProtonDelivery.class))); when(commandConsumerFactory.getCommandResponseSender(anyString(), anyString())) .thenReturn(Future.succeededFuture(sender)); final MqttServer server = getMqttServer(false); final AbstractVertxBasedMqttProtocolAdapter<MqttProtocolAdapterProperties> adapter = getAdapter(server); final MqttEndpoint endpoint = mockEndpoint(); when(endpoint.isConnected()).thenReturn(Boolean.TRUE); final MqttPublishMessage messageFromDevice = mock(MqttPublishMessage.class); when(messageFromDevice.qosLevel()).thenReturn(MqttQoS.AT_MOST_ONCE); when(messageFromDevice.messageId()).thenReturn(5555555); when(messageFromDevice.topicName()).thenReturn("command/my-tenant/4712/res/1010f8ab0b53-bd96-4d99-9d9c-56b868474a6a/200"); when(messageFromDevice.payload()).thenReturn(null); final ResourceIdentifier address = ResourceIdentifier .fromString("command/my-tenant/4712/res/1010f8ab0b53-bd96-4d99-9d9c-56b868474a6a/200"); adapter.uploadCommandResponseMessage(newMqttContext(messageFromDevice, endpoint, span), address) .onComplete(ctx.succeeding(result -> { ctx.verify(() -> { verify(sender).sendCommandResponse(any(), any()); verify(metrics).reportCommand( eq(MetricsTags.Direction.RESPONSE), eq("my-tenant"), any(TenantObject.class), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(0), any()); ctx.completeNow(); }); })); }