AdapterInstanceCommandHandler { public void handleCommandMessage(final Message msg, final ProtonDelivery delivery) { Objects.requireNonNull(msg); Objects.requireNonNull(delivery); final ResourceIdentifier resourceIdentifier = Strings.isNullOrEmpty(msg.getAddress()) ? null : ResourceIdentifier.fromString(msg.getAddress()); if (resourceIdentifier == null || resourceIdentifier.getResourceId() == null) { LOG.debug("address of command message is invalid: {}", msg.getAddress()); final Rejected rejected = new Rejected(); rejected.setError(new ErrorCondition(Constants.AMQP_BAD_REQUEST, "invalid command target address")); delivery.disposition(rejected, true); return; } final String tenantId = resourceIdentifier.getTenantId(); final String originalDeviceId = resourceIdentifier.getResourceId(); final String gatewayIdFromMessage = MessageHelper.getApplicationProperty(msg.getApplicationProperties(), MessageHelper.APP_PROPERTY_CMD_VIA, String.class); final String targetDeviceId = gatewayIdFromMessage != null ? gatewayIdFromMessage : originalDeviceId; final CommandHandlerWrapper commandHandler = getDeviceSpecificCommandHandler(tenantId, targetDeviceId); final String gatewayId = commandHandler != null && commandHandler.getGatewayId() != null ? commandHandler.getGatewayId() : gatewayIdFromMessage; final Command command = Command.from(msg, tenantId, gatewayId != null ? gatewayId : originalDeviceId); final SpanContext spanContext = TracingHelper.extractSpanContext(tracer, msg); final Span currentSpan = CommandConsumer.createSpan("handle command", tenantId, originalDeviceId, gatewayId, tracer, spanContext); currentSpan.setTag(MessageHelper.APP_PROPERTY_ADAPTER_INSTANCE_ID, adapterInstanceId); CommandConsumer.logReceivedCommandToSpan(command, currentSpan); if (commandHandler != null) { LOG.trace("using [{}] for received command [{}]", commandHandler, command); commandHandler.handleCommand(CommandContext.from(command, delivery, currentSpan)); } else { LOG.info("no command handler found for command with device id {}, gateway id {} [tenant-id: {}]", originalDeviceId, gatewayId, tenantId); TracingHelper.logError(currentSpan, "no command handler found for command"); currentSpan.finish(); ProtonHelper.released(delivery, true); } } AdapterInstanceCommandHandler(final Tracer tracer, final String adapterInstanceId); void handleCommandMessage(final Message msg, final ProtonDelivery delivery); CommandHandlerWrapper putDeviceSpecificCommandHandler(final String tenantId, final String deviceId,
            final String gatewayId, final Handler<CommandContext> commandHandler); CommandHandlerWrapper getDeviceSpecificCommandHandler(final String tenantId, final String deviceId); Collection<CommandHandlerWrapper> getDeviceSpecificCommandHandlers(); boolean removeDeviceSpecificCommandHandler(final String tenantId, final String deviceId);  }
@Test void testHandleCommandMessageWithInvalidMessage() { final Message msg = mock(Message.class); final ProtonDelivery delivery = mock(ProtonDelivery.class); adapterInstanceCommandHandler.handleCommandMessage(msg, delivery); final ArgumentCaptor<DeliveryState> deliveryStateCaptor = ArgumentCaptor.forClass(DeliveryState.class); verify(delivery).disposition(deliveryStateCaptor.capture(), anyBoolean()); assertThat(deliveryStateCaptor.getValue()).isNotNull(); assertThat(deliveryStateCaptor.getValue()).isInstanceOf(Rejected.class); }
@Test void testHandleCommandMessageWithNoHandlerFound() { final Message msg = mock(Message.class); final String deviceId = "4711"; when(msg.getAddress()).thenReturn(String.format("%s/%s/%s", CommandConstants.COMMAND_ENDPOINT, Constants.DEFAULT_TENANT, deviceId)); final ProtonDelivery delivery = mock(ProtonDelivery.class); adapterInstanceCommandHandler.handleCommandMessage(msg, delivery); final ArgumentCaptor<DeliveryState> deliveryStateCaptor = ArgumentCaptor.forClass(DeliveryState.class); verify(delivery).disposition(deliveryStateCaptor.capture(), anyBoolean()); assertThat(deliveryStateCaptor.getValue()).isNotNull(); assertThat(deliveryStateCaptor.getValue()).isInstanceOf(Released.class); }