AbstractVertxBasedHttpProtocolAdapter extends AbstractProtocolAdapterBase<T> { public final void uploadTelemetryMessage(final HttpContext ctx, final String tenant, final String deviceId) { uploadTelemetryMessage( Objects.requireNonNull(ctx), Objects.requireNonNull(tenant), Objects.requireNonNull(deviceId), ctx.getRoutingContext().getBody(), ctx.getContentType()); }  @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 testUploadTelemetryDoesNotWaitForAcceptedOutcome() { final Future<ProtonDelivery> outcome = Future.succeededFuture(mock(ProtonDelivery.class)); givenATelemetrySenderForOutcome(outcome); 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.uploadTelemetryMessage(ctx, "tenant", "device"); verify(response).setStatusCode(202); verify(response).end(); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_MOST_ONCE), eq(payload.length()), eq(MetricsTags.TtdStatus.NONE), any()); }
@Test public void testUploadTelemetryUsesConfiguredMaxTtd() { final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final DownstreamSender sender = givenATelemetrySenderForOutcome(Future.succeededFuture()); when(tenantClient.get(eq("tenant"), any())).thenReturn( Future.succeededFuture(TenantObject.from("tenant", true) .addAdapter(new Adapter(ADAPTER_TYPE).setEnabled(Boolean.TRUE) .setExtensions(Map.of(TenantConstants.FIELD_MAX_TTD, 20))))); final Buffer payload = Buffer.buffer("some payload"); final HttpServerResponse response = mock(HttpServerResponse.class); final HttpServerRequest request = mock(HttpServerRequest.class); when(request.getHeader(eq(Constants.HEADER_TIME_TILL_DISCONNECT))).thenReturn("40"); final HttpContext ctx = newHttpContext(payload, "application/text", request, response); adapter.uploadTelemetryMessage(ctx, "tenant", "device"); verify(response).setStatusCode(202); verify(response).end(); final ArgumentCaptor<Message> messageCaptor = ArgumentCaptor.forClass(Message.class); verify(sender).send(messageCaptor.capture(), (SpanContext) any()); assertThat(MessageHelper.getTimeUntilDisconnect(messageCaptor.getValue())).isEqualTo(20); }
@Test public void testMessageLimitExceededForATelemetryMessage() { final Future<ProtonDelivery> outcome = Future.succeededFuture(mock(ProtonDelivery.class)); givenATelemetrySenderForOutcome(outcome); final HttpServer server = getHttpServer(false); final AbstractVertxBasedHttpProtocolAdapter<HttpProtocolAdapterProperties> adapter = getAdapter(server, null); final Buffer payload = Buffer.buffer("some payload"); final HttpContext routingContext = newHttpContext(payload); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); adapter.uploadTelemetryMessage(routingContext, "my-tenant", "the-device", payload, "application/text"); assertContextFailedWithClientError(routingContext, HttpUtils.HTTP_TOO_MANY_REQUESTS); verify(metrics).reportTelemetry( eq(EndpointType.TELEMETRY), eq("my-tenant"), any(), eq(ProcessingOutcome.UNPROCESSABLE), eq(QoS.AT_MOST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }