AbstractVertxBasedCoapAdapter extends AbstractProtocolAdapterBase<T> { public final Future<ResponseCode> uploadEventMessage( final CoapContext context) { Objects.requireNonNull(context); if (context.isConfirmable()) { return doUploadMessage( context, true, getEventSender(context.getOriginDevice().getTenantId()), MetricsTags.EndpointType.EVENT); } else { context.respondWithCode(ResponseCode.BAD_REQUEST, "event endpoint supports confirmable request messages only"); return Future.succeededFuture(ResponseCode.BAD_REQUEST); } }  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 testUploadEventWaitsForAcceptedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenAnEventSender(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.uploadEventMessage(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.EVENT), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.FORWARDED), eq(MetricsTags.QoS.AT_LEAST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }
@Test public void testUploadEventFailsForRejectedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenAnEventSender(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 ctx = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadEventMessage(ctx); verify(sender).sendAndWaitForOutcome(any(Message.class), any(SpanContext.class)); outcome.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "malformed message")); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.BAD_REQUEST.equals(res.getCode()))); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.EVENT), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), eq(MetricsTags.QoS.AT_LEAST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }
@Test public void testMessageLimitExceededForAnEventMessage() { final Promise<ProtonDelivery> outcome = Promise.promise(); final DownstreamSender sender = givenAnEventSender(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.CON, MediaTypeRegistry.TEXT_PLAIN); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext ctx = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadEventMessage(ctx); verify(sender, never()).sendAndWaitForOutcome(any(Message.class), any(SpanContext.class)); final ArgumentCaptor<Response> captor = ArgumentCaptor.forClass(Response.class); verify(coapExchange).respond(captor.capture()); assertThat(captor.getValue().getCode()).isEqualTo(ResponseCode.TOO_MANY_REQUESTS); verify(metrics).reportTelemetry( eq(MetricsTags.EndpointType.EVENT), eq("tenant"), any(), eq(MetricsTags.ProcessingOutcome.UNPROCESSABLE), eq(MetricsTags.QoS.AT_LEAST_ONCE), eq(payload.length()), eq(TtdStatus.NONE), any()); }