VertxBasedAmqpProtocolAdapter extends AbstractProtocolAdapterBase<AmqpAdapterProperties> { protected void onCommandReceived(final TenantObject tenantObject, final ProtonSender sender, final CommandContext commandContext) { Objects.requireNonNull(tenantObject); Objects.requireNonNull(sender); Objects.requireNonNull(commandContext); final Command command = commandContext.getCommand(); final Message msg = command.getCommandMessage(); if (msg.getCorrelationId() == null) { msg.setCorrelationId(msg.getMessageId()); } if (command.isTargetedAtGateway()) { MessageHelper.addDeviceId(msg, command.getOriginalDeviceId()); } final AtomicBoolean isCommandSettled = new AtomicBoolean(false); if (sender.sendQueueFull()) { log.debug("cannot send command to device: no credit available [{}]", command); final Exception ex = new ServerErrorException(HttpURLConnection.HTTP_UNAVAILABLE, "no credit available for sending command to device"); TracingHelper.logError(commandContext.getTracingSpan(), ex); commandContext.release(); reportSentCommand(tenantObject, commandContext, ProcessingOutcome.UNDELIVERABLE); } else { final Long timerId = getConfig().getSendMessageToDeviceTimeout() < 1 ? null : vertx.setTimer(getConfig().getSendMessageToDeviceTimeout(), tid -> { log.debug("waiting for delivery update timed out after {}ms [{}]", getConfig().getSendMessageToDeviceTimeout(), command); if (isCommandSettled.compareAndSet(false, true)) { final Exception ex = new ServerErrorException(HttpURLConnection.HTTP_UNAVAILABLE, "timeout waiting for delivery update from device"); TracingHelper.logError(commandContext.getTracingSpan(), ex); commandContext.release(); reportSentCommand(tenantObject, commandContext, ProcessingOutcome.UNDELIVERABLE); } else { log.trace("command is already settled and downstream application was already notified [{}]", command); } }); sender.send(msg, delivery -> { if (timerId != null) { vertx.cancelTimer(timerId); } if (!isCommandSettled.compareAndSet(false, true)) { log.trace("command is already settled and downstream application was already notified [{}]", command); } else { final DeliveryState remoteState = delivery.getRemoteState(); ProcessingOutcome outcome = null; if (delivery.remotelySettled()) { commandContext.disposition(remoteState); if (Accepted.class.isInstance(remoteState)) { outcome = ProcessingOutcome.FORWARDED; } else if (Rejected.class.isInstance(remoteState)) { outcome = ProcessingOutcome.UNPROCESSABLE; } else if (Released.class.isInstance(remoteState)) { outcome = ProcessingOutcome.UNDELIVERABLE; } else if (Modified.class.isInstance(remoteState)) { final Modified modified = (Modified) remoteState; outcome = modified.getUndeliverableHere() ? ProcessingOutcome.UNPROCESSABLE : ProcessingOutcome.UNDELIVERABLE; } } else { log.debug("device did not settle command message [remote state: {}, {}]", remoteState, command); final Map<String, Object> logItems = new HashMap<>(2); logItems.put(Fields.EVENT, "device did not settle command"); logItems.put("remote state", remoteState); commandContext.getTracingSpan().log(logItems); commandContext.release(); outcome = ProcessingOutcome.UNDELIVERABLE; } reportSentCommand(tenantObject, commandContext, outcome); } }); final Map<String, Object> items = new HashMap<>(4); items.put(Fields.EVENT, "command sent to device"); if (sender.getRemoteTarget() != null) { items.put(Tags.MESSAGE_BUS_DESTINATION.getKey(), sender.getRemoteTarget().getAddress()); } items.put(TracingHelper.TAG_QOS.getKey(), sender.getQoS().name()); items.put(TracingHelper.TAG_CREDIT.getKey(), sender.getCredit()); commandContext.getTracingSpan().log(items); } }  @Autowired void setMetrics(final AmqpAdapterMetrics metrics); @Override int getPortDefaultValue(); @Override int getInsecurePortDefaultValue();  }
@SuppressWarnings("unchecked") @Test public void testLinkForSendingCommandsCloseAfterTimeout() { final VertxBasedAmqpProtocolAdapter adapter = givenAnAmqpAdapter(); final ProtonSender deviceLink = mock(ProtonSender.class); when(deviceLink.getQoS()).thenReturn(ProtonQoS.AT_LEAST_ONCE); final TenantObject tenantObject = givenAConfiguredTenant(TEST_TENANT_ID, true); final ProtonDelivery commandDelivery = mock(ProtonDelivery.class); final String commandAddress = String.format("%s/%s/%s", getCommandEndpoint(), TEST_TENANT_ID, TEST_DEVICE); final Buffer payload = Buffer.buffer("payload"); final Message message = getFakeMessage(commandAddress, payload, "commandToExecute"); final Command command = Command.from(message, TEST_TENANT_ID, TEST_DEVICE); final CommandContext context = CommandContext.from(command, commandDelivery, mock(Span.class)); final ProtocolAdapterCommandConsumer commandConsumer = mock(ProtocolAdapterCommandConsumer.class); when(commandConsumer.close(any())).thenReturn(Future.succeededFuture()); doAnswer(invocation -> { final Handler<Long> task = invocation.getArgument(1); task.handle(1L); return 1L; }).when(vertx).setTimer(anyLong(), any(Handler.class)); adapter.onCommandReceived(tenantObject, deviceLink, context); verify(commandDelivery).disposition(any(Released.class), eq(true)); }