AbstractVertxBasedCoapAdapter extends AbstractProtocolAdapterBase<T> { public final Future<ResponseCode> uploadTelemetryMessage( final CoapContext context) { return doUploadMessage( Objects.requireNonNull(context), context.isConfirmable(), getTelemetrySender(context.getOriginDevice().getTenantId()), MetricsTags.EndpointType.TELEMETRY); }  final void setHonoDeviceResolver(final ApplicationLevelInfoSupplier deviceIdentityResolver); final void setPskStore(final AdvancedPskStore pskStore); @Autowired final void setMetrics(final CoapAdapterMetrics metrics); @Autowired(required = false) final void setResources(final Set<Resource> resources); @Override final int getPortDefaultValue(); @Override final int getInsecurePortDefaultValue(); @Autowired(required = false) final void setCoapServer(final CoapServer server); @Override final void doStart(final Promise<Void> startPromise); @Override final void doStop(final Promise<Void> stopPromise); final Future<ResponseCode> uploadTelemetryMessage(
            final CoapContext context); final Future<ResponseCode> uploadEventMessage(
            final CoapContext context); final Future<ResponseCode> uploadCommandResponseMessage(final CoapContext context);  }
@Test public void testUploadTelemetryFailsForDisabledTenant() { final DownstreamSender sender = givenATelemetrySender(Promise.promise()); final TenantObject myTenantConfig = TenantObject.from("my-tenant", true); myTenantConfig.addAdapter(new Adapter(ADAPTER_TYPE).setEnabled(Boolean.FALSE)); when(tenantClient.get(eq("my-tenant"), any(SpanContext.class))).thenReturn(Future.succeededFuture(myTenantConfig)); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final Buffer payload = Buffer.buffer("some payload"); final CoapExchange coapExchange = newCoapExchange(payload, Type.NON, MediaTypeRegistry.TEXT_PLAIN); final Device authenticatedDevice = new Device("my-tenant", "the-device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "the-device", span); adapter.uploadTelemetryMessage(context); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.FORBIDDEN.equals(res.getCode()))); verify(sender, never()).send(any(Message.class), any(SpanContext.class)); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("my-tenant"), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), eq(MetricsTags.QoS.AT_MOST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }
@Test public void testUploadTelemetryFailsForMissingContentFormat() { final DownstreamSender sender = givenATelemetrySender(Promise.promise()); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final Buffer payload = Buffer.buffer("some payload"); final CoapExchange coapExchange = newCoapExchange(payload, Type.NON, (Integer) null); final Device authenticatedDevice = new Device("my-tenant", "the-device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "the-device", span); adapter.uploadTelemetryMessage(context); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.BAD_REQUEST.equals(res.getCode()))); verify(sender, never()).send(any(Message.class), any(SpanContext.class)); verify(metrics, never()).reportTelemetry( any(MetricsTags.EndpointType.class), anyString(), any(), any(MetricsTags.ProcessingOutcome.class), any(MetricsTags.QoS.class), anyInt(), any(TtdStatus.class), any()); }
@Test public void testUploadTelemetryFailsForEmptyBody() { final DownstreamSender sender = givenATelemetrySender(Promise.promise()); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final CoapExchange coapExchange = newCoapExchange(null, Type.NON, MediaTypeRegistry.UNDEFINED); final Device authenticatedDevice = new Device("my-tenant", "the-device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "the-device", span); adapter.uploadTelemetryMessage(context); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.BAD_REQUEST.equals(res.getCode()))); verify(sender, never()).send(any(Message.class), any(SpanContext.class)); verify(metrics, never()).reportTelemetry( any(MetricsTags.EndpointType.class), anyString(), any(), any(MetricsTags.ProcessingOutcome.class), any(MetricsTags.QoS.class), anyInt(), any(TtdStatus.class), any()); }
@Test public void testUploadEmptyNotificationSucceeds() { final DownstreamSender sender = givenATelemetrySender(Promise.promise()); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final OptionSet options = new OptionSet(); options.addUriQuery(CoapContext.PARAM_EMPTY_CONTENT); final CoapExchange coapExchange = newCoapExchange(null, Type.NON, options); final Device authenticatedDevice = new Device("my-tenant", "the-device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "the-device", span); adapter.uploadTelemetryMessage(context); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.CHANGED.equals(res.getCode()))); verify(sender).send(argThat(msg -> EventConstants.isEmptyNotificationType(msg.getContentType())), any(SpanContext.class)); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("my-tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_MOST_ONCE), eq(0), eq(TtdStatus.NONE), any()); }
@Test public void testUploadTelemetryWithQoS0() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenATelemetrySender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final Buffer payload = Buffer.buffer("some payload"); final CoapExchange coapExchange = newCoapExchange(payload, Type.NON, MediaTypeRegistry.TEXT_PLAIN); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadTelemetryMessage(context); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.CHANGED.equals(res.getCode()))); verify(sender).send(any(Message.class), any(SpanContext.class)); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_MOST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }
@Test public void testUploadTelemetryWithQoS1() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenATelemetrySender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final Buffer payload = Buffer.buffer("some payload"); final CoapExchange coapExchange = newCoapExchange(payload, Type.CON, MediaTypeRegistry.TEXT_PLAIN); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadTelemetryMessage(context); verify(sender).sendAndWaitForOutcome(any(Message.class), any(SpanContext.class)); verify(coapExchange, never()).respond(any(Response.class)); outcome.complete(mock(ProtonDelivery.class)); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.CHANGED.equals(res.getCode()))); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_LEAST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }
@Test public void testMessageLimitExceededForATelemetryMessage() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenATelemetrySender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); when(resourceLimitChecks.isMessageLimitReached(any(TenantObject.class), anyLong(), any(SpanContext.class))) .thenReturn(Future.succeededFuture(Boolean.TRUE)); final Buffer payload = Buffer.buffer("some payload"); final CoapExchange coapExchange = newCoapExchange(payload, Type.NON, MediaTypeRegistry.TEXT_PLAIN); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext ctx = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadTelemetryMessage(ctx); verify(sender, never()).send(any(Message.class), any(SpanContext.class)); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.TOO_MANY_REQUESTS.equals(res.getCode()))); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.TELEMETRY), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), eq(MetricsTags.QoS.AT_MOST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }