AbstractVertxBasedCoapAdapter extends AbstractProtocolAdapterBase<T> { public final Future<ResponseCode> uploadCommandResponseMessage(final CoapContext context) { Objects.requireNonNull(context); final Device device = context.getOriginDevice(); final Device authenticatedDevice = context.getAuthenticatedDevice(); if (!context.isConfirmable()) { context.respondWithCode(ResponseCode.BAD_REQUEST, "command response endpoint supports confirmable request messages only"); return Future.succeededFuture(ResponseCode.BAD_REQUEST); } final Buffer payload = context.getPayload(); final String contentType = context.getContentType(); final String commandRequestId = context.getCommandRequestId(); final Integer responseStatus = context.getCommandResponseStatus(); log.debug("processing response to command [tenantId: {}, deviceId: {}, cmd-req-id: {}, status code: {}]", device.getTenantId(), device.getDeviceId(), commandRequestId, responseStatus); final Span currentSpan = TracingHelper .buildChildSpan(tracer, context.getTracingContext(), "upload Command response", getTypeName()) .withTag(Tags.SPAN_KIND.getKey(), Tags.SPAN_KIND_CLIENT) .withTag(TracingHelper.TAG_TENANT_ID, device.getTenantId()) .withTag(TracingHelper.TAG_DEVICE_ID, device.getDeviceId()) .withTag(Constants.HEADER_COMMAND_RESPONSE_STATUS, responseStatus) .withTag(Constants.HEADER_COMMAND_REQUEST_ID, commandRequestId) .withTag(TracingHelper.TAG_AUTHENTICATED.getKey(), authenticatedDevice != null) .start(); final Future<TenantObject> tenantTracker = getTenantConfiguration(device.getTenantId(), currentSpan.context()); final Optional<CommandResponse> cmdResponse = Optional.ofNullable(CommandResponse.from( commandRequestId, device.getTenantId(), device.getDeviceId(), payload, contentType, responseStatus)); final Future<CommandResponse> commandResponseTracker = cmdResponse .map(res -> Future.succeededFuture(res)) .orElseGet(() -> Future.failedFuture( new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, String.format("command-request-id [%s] or status code [%s] is missing/invalid", commandRequestId, responseStatus)))); return CompositeFuture.all(tenantTracker, commandResponseTracker) .compose(ok -> { final Future<JsonObject> deviceRegistrationTracker = getRegistrationAssertion( device.getTenantId(), device.getDeviceId(), authenticatedDevice, currentSpan.context()); final Future<Void> tenantValidationTracker = CompositeFuture.all( isAdapterEnabled(tenantTracker.result()), checkMessageLimit(tenantTracker.result(), payload.length(), currentSpan.context())) .mapEmpty(); return CompositeFuture.all(tenantValidationTracker, deviceRegistrationTracker); }) .compose(ok -> sendCommandResponse(device.getTenantId(), 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, device.getTenantId(), tenantTracker.result(), ProcessingOutcome.FORWARDED, payload.length(), context.getTimer()); context.respondWithCode(ResponseCode.CHANGED); return ResponseCode.CHANGED; }) .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, device.getTenantId(), tenantTracker.result(), ProcessingOutcome.from(t), payload.length(), context.getTimer()); final Response response = CoapErrorResponse.respond(t, ResponseCode.INTERNAL_SERVER_ERROR); return context.respond(response); }); }  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 testUploadCommandResponseWaitsForAcceptedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); final CommandResponseSender sender = givenACommandResponseSender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final String reqId = Command.getRequestId("correlation", "replyToId", "device"); final Buffer payload = Buffer.buffer("some payload"); final OptionSet options = new OptionSet(); options.addUriPath(CommandConstants.COMMAND_RESPONSE_ENDPOINT).addUriPath(reqId); options.addUriQuery(String.format("%s=%d", Constants.HEADER_COMMAND_RESPONSE_STATUS, 200)); options.setContentFormat(MediaTypeRegistry.TEXT_PLAIN); final CoapExchange coapExchange = newCoapExchange(payload, Type.CON, options); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadCommandResponseMessage(context); verify(sender).sendCommandResponse(any(CommandResponse.class), any(SpanContext.class)); verify(coapExchange, never()).respond(any(Response.class)); verify(metrics, never()).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), anyInt(), any()); outcome.complete(mock(ProtonDelivery.class)); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.CHANGED.equals(res.getCode()))); verify(metrics).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), eq(payload.length()), any()); }
@Test public void testUploadCommandResponseFailsForRejectedOutcome() { final Promise<ProtonDelivery> outcome = Promise.promise(); final CommandResponseSender sender = givenACommandResponseSender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final String reqId = Command.getRequestId("correlation", "replyToId", "device"); final Buffer payload = Buffer.buffer("some payload"); final OptionSet options = new OptionSet(); options.addUriPath(CommandConstants.COMMAND_RESPONSE_ENDPOINT).addUriPath(reqId); options.addUriQuery(String.format("%s=%d", Constants.HEADER_COMMAND_RESPONSE_STATUS, 200)); options.setContentFormat(MediaTypeRegistry.TEXT_PLAIN); final CoapExchange coapExchange = newCoapExchange(payload, Type.CON, options); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadCommandResponseMessage(context); outcome.fail(new ClientErrorException(HttpURLConnection.HTTP_BAD_REQUEST, "malformed message")); verify(sender).sendCommandResponse(any(CommandResponse.class), any(SpanContext.class)); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.BAD_REQUEST.equals(res.getCode()))); verify(metrics, never()).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), anyInt(), any()); }
@Test public void testUploadCommandResponseFailsForDisabledTenant() { final TenantObject to = TenantObject.from("tenant", true); to.addAdapter(new Adapter(Constants.PROTOCOL_ADAPTER_TYPE_COAP).setEnabled(Boolean.FALSE)); when(tenantClient.get(eq("tenant"), (SpanContext) any())).thenReturn(Future.succeededFuture(to)); final Promise<ProtonDelivery> outcome = Promise.promise(); final CommandResponseSender sender = givenACommandResponseSender(outcome); final CoapServer server = getCoapServer(false); final AbstractVertxBasedCoapAdapter<CoapAdapterProperties> adapter = getAdapter(server, true, null); final String reqId = Command.getRequestId("correlation", "replyToId", "device"); final Buffer payload = Buffer.buffer("some payload"); final OptionSet options = new OptionSet(); options.addUriPath(CommandConstants.COMMAND_RESPONSE_ENDPOINT).addUriPath(reqId); options.addUriQuery(String.format("%s=%d", Constants.HEADER_COMMAND_RESPONSE_STATUS, 200)); options.setContentFormat(MediaTypeRegistry.TEXT_PLAIN); final CoapExchange coapExchange = newCoapExchange(payload, Type.CON, options); final Device authenticatedDevice = new Device("tenant", "device"); final CoapContext context = CoapContext.fromRequest(coapExchange, authenticatedDevice, authenticatedDevice, "device", span); adapter.uploadCommandResponseMessage(context); verify(sender, never()).sendCommandResponse(any(CommandResponse.class), any(SpanContext.class)); verify(coapExchange).respond(argThat((Response res) -> ResponseCode.FORBIDDEN.equals(res.getCode()))); verify(metrics, never()).reportCommand( eq(Direction.RESPONSE), eq("tenant"), any(), eq(ProcessingOutcome.FORWARDED), anyInt(), any()); }