AbstractVertxBasedHttpProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final void uploadCommandResponseMessage( final HttpContext ctx, final String tenant, final String deviceId, final String commandRequestId, final Integer responseStatus) { Objects.requireNonNull(ctx); Objects.requireNonNull(tenant); Objects.requireNonNull(deviceId); final Buffer payload = ctx.getRoutingContext().getBody(); final String contentType = ctx.getContentType(); log.debug("processing response to command [tenantId: {}, deviceId: {}, cmd-req-id: {}, status code: {}]", tenant, deviceId, commandRequestId, responseStatus); final Device authenticatedDevice = ctx.getAuthenticatedDevice(); final Span currentSpan = TracingHelper .buildChildSpan(tracer, TracingHandler.serverSpanContext(ctx.getRoutingContext()), "upload Command response", getTypeName()) .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT) .withTag(TracingHelper.TAG_TENANT_ID, tenant) .withTag(TracingHelper.TAG_DEVICE_ID, deviceId) .withTag(Constants.HEADER_COMMAND_RESPONSE_STATUS, responseStatus) .withTag(Constants.HEADER_COMMAND_REQUEST_ID, commandRequestId) .withTag(TracingHelper.TAG_AUTHENTICATED.getKey(), authenticatedDevice != null) .start(); final CommandResponse cmdResponseOrNull = CommandResponse.from(commandRequestId, tenant, deviceId, payload, contentType, responseStatus); final Future<TenantObject> tenantTracker = getTenantConfiguration(tenant, currentSpan.context()); final Future<CommandResponse> commandResponseTracker = cmdResponseOrNull != null ? Future.succeededFuture(cmdResponseOrNull) : Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, String.format("command-request-id [%s] or status code [%s] is missing/invalid", commandRequestId, responseStatus))); final int payloadSize = Optional.ofNullable(payload).map(Buffer::length).orElse(0); CompositeFuture.all(tenantTracker, commandResponseTracker) .compose(commandResponse -> { final Future<JsonObject> deviceRegistrationTracker = getRegistrationAssertion( tenant, deviceId, authenticatedDevice, currentSpan.context()); final Future<Void> tenantValidationTracker = CompositeFuture .all(isAdapterEnabled(tenantTracker.result()), checkMessageLimit(tenantTracker.result(), payloadSize, currentSpan.context())) .map(ok -> null); return CompositeFuture.all(tenantValidationTracker, deviceRegistrationTracker) .compose(ok -> sendCommandResponse(tenant, commandResponseTracker.result(), currentSpan.context())) .map(delivery -> { log.trace("delivered command response [command-request-id: {}] to application", commandRequestId); currentSpan.log("delivered command response to application"); currentSpan.finish(); metrics.reportCommand( Direction.RESPONSE, tenant, tenantTracker.result(), ProcessingOutcome.FORWARDED, payloadSize, getMicrometerSample(ctx.getRoutingContext())); ctx.response().setStatusCode(HttpURLConnection.HTTP_ACCEPTED); ctx.response().end(); return delivery; }); }).otherwise(t -> { log.debug("could not send command response [command-request-id: {}] to application", commandRequestId, t); TracingHelper.logError(currentSpan, t); currentSpan.finish(); metrics.reportCommand( Direction.RESPONSE, tenant, tenantTracker.result(), ProcessingOutcome.from(t), payloadSize, getMicrometerSample(ctx.getRoutingContext())); ctx.fail(t); return null; }); }  @Autowired final void setMetrics(final HttpAdapterMetrics metrics); @Override final int getPortDefaultValue(); @Override final int getInsecurePortDefaultValue(); @Autowired(required = false) final void setHttpServer(final HttpServer server); @Autowired(required = false) final void setInsecureHttpServer(final HttpServer server); @Override final void doStart(final Promise<Void> startPromise); @Override final void doStop(final Promise<Void> stopPromise); final void uploadTelemetryMessage(final HttpContext ctx, final String tenant, final String deviceId); final void uploadTelemetryMessage(final HttpContext ctx, final String tenant, final String deviceId,
            final Buffer payload, final String contentType); final void uploadEventMessage(final HttpContext ctx, final String tenant, final String deviceId); final void uploadEventMessage(final HttpContext ctx, final String tenant, final String deviceId,
            final Buffer payload, final String contentType); final void uploadCommandResponseMessage(
            final HttpContext ctx,
            final String tenant,
            final String deviceId,
            final String commandRequestId,
            final Integer responseStatus);  }
@SuppressWarnings("unchecked") @Test public void testUploadCommandResponseWaitsForAcceptedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); givenACommandResponseSenderForOutcome(outcome.future()); final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = Buffer.buffer("some payload"); final HttpServerResponse response = mock(HttpServerResponse.class); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), response); when(ctx.getRoutingContext().addBodyEndHandler(any(Handler.class))).thenAnswer(invocation -> { final Handler<Void> handler = invocation.getArgument(0); handler.handle(null); return 0; }); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); verify(response, never()).end(); verify(metrics, never()).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), anyInt(), any()); outcome.complete(mock(ProtonDelivery.class)); verify(response).setStatusCode(202); verify(response).end(); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), eq(payload.length()), any()); }
@SuppressWarnings("unchecked") @Test public void testUploadEmptyCommandResponseSucceeds() { 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 HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = null; final HttpServerResponse response = mock(HttpServerResponse.class); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), response); when(ctx.getRoutingContext().addBodyEndHandler(any(Handler.class))).thenAnswer(invocation -> { final Handler<Void> handler = invocation.getArgument(0); handler.handle(null); return 0; }); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); verify(response).setStatusCode(202); verify(response).end(); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), eq(0), any()); }
@Test public void testUploadCommandResponseFailsForDisabledTenant() { final TenantObject to = TenantObject.from("tenant", true); to.addAdapter(new Adapter(Constants.PROTOCOL_ADAPTER_TYPE_HTTP).setEnabled(Boolean.FALSE)); when(tenantClient.get(eq("tenant"), (SpanContext) any())).thenReturn(Future.succeededFuture(to)); final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = Buffer.buffer("some payload"); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), mock(HttpServerResponse.class)); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); assertContextFailedWithClientError(ctx, HttpURLConnection.HTTP_FORBIDDEN); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), eq(to), eq(ProcessingOutcome.UNPROCESSABLE), eq(payload.length()), any()); }
@Test public void testUploadCommandResponseFailsForOtherDevice() { final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = Buffer.buffer("some payload"); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), mock(HttpServerResponse.class)); final TenantObject to = TenantObject.from("tenant", true); to.addAdapter(new Adapter(ADAPTER_TYPE).setEnabled(Boolean.TRUE)); when(tenantClient.get(eq("tenant"), (SpanContext) any())).thenReturn(Future.succeededFuture(to)); when(ctx.getAuthenticatedDevice()).thenReturn(new DeviceUser("tenant", "9999")); when(regClient.assertRegistration((String) any(), (String) any(), (SpanContext) any())) .thenReturn(Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_NOT_FOUND, "cannot publish data for device of other tenant"))); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); assertContextFailedWithClientError(ctx, HttpURLConnection.HTTP_NOT_FOUND); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), eq(to), eq(ProcessingOutcome.UNPROCESSABLE), eq(payload.length()), any()); }
@Test public void testUploadCommandResponseFailsForRejectedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); givenACommandResponseSenderForOutcome(outcome.future()); final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = Buffer.buffer("some payload"); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), mock(HttpServerResponse.class)); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); outcome.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "malformed message")); assertContextFailedWithClientError(ctx, HttpURLConnection.HTTP_BAD_REQUEST); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.UNPROCESSABLE), eq(payload.length()), any()); }
@SuppressWarnings("unchecked") @Test public void testMessageLimitExceededForACommandResponseMessage() { final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter( getHttpServer(false), null); final Promise<ProtonDelivery> outcome = Promise.promise(); givenACommandResponseSenderForOutcome(outcome.future()); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final Buffer payload = Buffer.buffer("some payload"); final HttpServerResponse response = mock(HttpServerResponse.class); final HttpContext ctx = newHttpContext(payload, "application/text", mock(HttpServerRequest.class), response); when(ctx.getRoutingContext().addBodyEndHandler(any(Handler.class))).thenAnswer(invocation -> { final Handler<Void> handler = invocation.getArgument(0); handler.handle(null); return 0; }); adapter.uploadCommandResponseMessage(ctx, "tenant", "device", CMD_REQ_ID, 200); assertContextFailedWithClientError(ctx, HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.UNPROCESSABLE), eq(payload.length()), any()); }