MappingAndDelegatingCommandHandler { public void mapAndDelegateIncomingCommandMessage( final String tenantId, final ProtonDelivery originalMessageDelivery, final Message message) { Objects.requireNonNull(tenantId); Objects.requireNonNull(originalMessageDelivery); Objects.requireNonNull(message); if (Strings.isNullOrEmpty(message.getAddress())) { LOG.debug("command message has no address"); final Rejected rejected = new Rejected(); rejected.setError(new ErrorCondition(Constants.AMQP_BAD_REQUEST, "missing command target address")); originalMessageDelivery.disposition(rejected, true); return; } final ResourceIdentifier targetAddress = ResourceIdentifier.fromString(message.getAddress()); final String deviceId = targetAddress.getResourceId(); if (!tenantId.equals(targetAddress.getTenantId())) { LOG.debug("command message address contains invalid tenant [expected: {}, found: {}]", tenantId, targetAddress.getTenantId()); final Rejected rejected = new Rejected(); rejected.setError(new ErrorCondition(AmqpError.UNAUTHORIZED_ACCESS, "unauthorized to send command to tenant")); originalMessageDelivery.disposition(rejected, true); return; } else if (deviceId == null) { LOG.debug("invalid command message address: {}", message.getAddress()); final Rejected rejected = new Rejected(); rejected.setError(new ErrorCondition(Constants.AMQP_BAD_REQUEST, "invalid command target address")); originalMessageDelivery.disposition(rejected, true); return; } final Command command = Command.from(message, tenantId, deviceId); if (command.isValid()) { LOG.trace("received valid command message: [{}]", command); } else { LOG.debug("received invalid command message: {}", command); } final SpanContext spanContext = TracingHelper.extractSpanContext(connection.getTracer(), message); final Span currentSpan = CommandConsumer.createSpan("map and delegate command", tenantId, deviceId, null, connection.getTracer(), spanContext); CommandConsumer.logReceivedCommandToSpan(command, currentSpan); final CommandContext commandContext = CommandContext.from(command, originalMessageDelivery, currentSpan); mapAndDelegateIncomingCommand(tenantId, deviceId, commandContext); } MappingAndDelegatingCommandHandler(
            final HonoConnection connection,
            final CommandTargetMapper commandTargetMapper,
            final AdapterInstanceCommandHandler adapterInstanceCommandHandler,
            final String adapterInstanceId,
            final SendMessageSampler sampler); void mapAndDelegateIncomingCommandMessage(
            final String tenantId,
            final ProtonDelivery originalMessageDelivery,
            final Message message);  }
@Test public void testMapWithLocalCommandHandler() { final String deviceId = "4711"; final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, null, localHandlerCmdContextRef::set); when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); assertThat(localHandlerCmdContextRef.get()).isNotNull(); assertThat(localHandlerCmdContextRef.get().getCommand()).isNotNull(); assertThat(localHandlerCmdContextRef.get().getCommand().getCommandMessage()).isEqualTo(message); assertThat(localHandlerCmdContextRef.get().getCommand().isValid()).isTrue(); verify(delivery, never()).disposition(any(), anyBoolean()); }
@Test public void testMapWithLocalCommandHandlerAndInvalidMessage() { final String deviceId = "4711"; final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, null, localHandlerCmdContextRef::set); when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); message.setSubject(null); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); assertThat(localHandlerCmdContextRef.get()).isNotNull(); assertThat(localHandlerCmdContextRef.get().getCommand()).isNotNull(); assertThat(localHandlerCmdContextRef.get().getCommand().getCommandMessage()).isEqualTo(message); assertThat(localHandlerCmdContextRef.get().getCommand().isValid()).isFalse(); verify(delivery, never()).disposition(any(), anyBoolean()); }
@SuppressWarnings("unchecked") @Test public void testMapForMessageHavingAddressWithoutDeviceId() { final String deviceId = "4711"; final Message message = getValidCommandMessage(deviceId); message.setAddress(String.format("%s/%s", CommandConstants.COMMAND_ENDPOINT, Constants.DEFAULT_TENANT)); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition( argThat(state -> Constants.AMQP_BAD_REQUEST.equals(((Rejected) state).getError().getCondition())), eq(true)); verify(sender, never()).send(any(Message.class), any(Handler.class)); }
@SuppressWarnings("unchecked") @Test public void testMapForMessageHavingAddressWithInvalidTenant() { final String deviceId = "4711"; final Message message = getValidCommandMessage(deviceId); message.setAddress(String.format("%s/%s/%s", CommandConstants.COMMAND_ENDPOINT, "wrong-tenant", deviceId)); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition( argThat(state -> AmqpError.UNAUTHORIZED_ACCESS.equals(((Rejected) state).getError().getCondition())), eq(true)); verify(sender, never()).send(any(Message.class), any(Handler.class)); }
@Test public void testMapWithNoAdapterInstanceFound() { final String deviceId = "4711"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Released.class), eq(true)); }
@Test public void testMapWithNoAdapterInstanceFoundAndMessageInvalid() { final String deviceId = "4711"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); message.setSubject(null); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Rejected.class), eq(true)); }
@Test public void testMapWithCommandHandlerNotFound() { final String deviceId = "4711"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.failedFuture(new ClientErrorException(HttpURLConnection.HTTP_NOT_FOUND))); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Released.class), eq(true)); }
@Test public void testMapWithLocalCommandHandlerForGateway() { final String deviceId = "4711"; final String gatewayId = "gw-1"; final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, gatewayId, null, localHandlerCmdContextRef::set); when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(gatewayId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); assertThat(localHandlerCmdContextRef.get()).isNotNull(); final Command command = localHandlerCmdContextRef.get().getCommand(); assertThat(command).isNotNull(); assertThat(command.isValid()).isTrue(); assertThat(command.getDeviceId()).isEqualTo(gatewayId); assertThat(command.getOriginalDeviceId()).isEqualTo(deviceId); verify(delivery, never()).disposition(any(), anyBoolean()); }
@Test public void testMapWithLocalCommandHandlerForGatewayAndSpecificDevice() { final String deviceId = "4711"; final String gatewayId = "gw-1"; final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, gatewayId, localHandlerCmdContextRef::set); when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, adapterInstanceId))); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); assertThat(localHandlerCmdContextRef.get()).isNotNull(); final Command command = localHandlerCmdContextRef.get().getCommand(); assertThat(command).isNotNull(); assertThat(command.isValid()).isTrue(); assertThat(command.getDeviceId()).isEqualTo(gatewayId); assertThat(command.getOriginalDeviceId()).isEqualTo(deviceId); verify(delivery, never()).disposition(any(), anyBoolean()); }
@Test public void testMapWithCommandHandlerOnAnotherInstance() { final String deviceId = "4711"; final String otherAdapterInstance = "otherAdapterInstance"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, otherAdapterInstance))); final ProtonDelivery sendMsgDeliveryUpdate = mock(ProtonDelivery.class); when(sendMsgDeliveryUpdate.getRemoteState()).thenReturn(new Accepted()); when(sendMsgDeliveryUpdate.remotelySettled()).thenReturn(true); final AtomicReference<Message> delegatedMessageRef = new AtomicReference<>(); when(sender.send(any(Message.class), VertxMockSupport.anyHandler())).thenAnswer(invocation -> { delegatedMessageRef.set(invocation.getArgument(0)); final Handler<ProtonDelivery> dispositionHandler = invocation.getArgument(1); dispositionHandler.handle(sendMsgDeliveryUpdate); return mock(ProtonDelivery.class); }); final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, null, localHandlerCmdContextRef::set); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Accepted.class), eq(true)); assertThat(localHandlerCmdContextRef.get()).isNull(); final Message delegatedMessage = delegatedMessageRef.get(); assertThat(delegatedMessage).isNotNull(); assertThat(delegatedMessage.getAddress()).isEqualTo(String.format("%s/%s/%s", CommandConstants.COMMAND_ENDPOINT, Constants.DEFAULT_TENANT, deviceId)); }
@Test public void testMapWithCommandHandlerOnAnotherInstanceWithInvalidMessage() { final String deviceId = "4711"; final String otherAdapterInstance = "otherAdapterInstance"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, otherAdapterInstance))); final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, null, localHandlerCmdContextRef::set); final Message message = getValidCommandMessage(deviceId); message.setSubject(null); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Rejected.class), eq(true)); assertThat(localHandlerCmdContextRef.get()).isNull(); }
@Test public void testMapWithCommandHandlerOnAnotherInstanceWithMessageSendingFailed() { final String deviceId = "4711"; final String otherAdapterInstance = "otherAdapterInstance"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(deviceId, otherAdapterInstance))); when(sender.sendQueueFull()).thenReturn(Boolean.TRUE); final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, deviceId, null, localHandlerCmdContextRef::set); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Released.class), eq(true)); assertThat(localHandlerCmdContextRef.get()).isNull(); }
@Test public void testMapWithCommandHandlerForGatewayOnAnotherInstance() { final String deviceId = "4711"; final String gatewayId = "gw-1"; final String otherAdapterInstance = "otherAdapterInstance"; when(commandTargetMapper.getTargetGatewayAndAdapterInstance(anyString(), anyString(), any())) .thenReturn(Future.succeededFuture(createTargetAdapterInstanceJson(gatewayId, otherAdapterInstance))); final ProtonDelivery sendMsgDeliveryUpdate = mock(ProtonDelivery.class); when(sendMsgDeliveryUpdate.getRemoteState()).thenReturn(new Accepted()); when(sendMsgDeliveryUpdate.remotelySettled()).thenReturn(true); final AtomicReference<Message> delegatedMessageRef = new AtomicReference<>(); when(sender.send(any(Message.class), VertxMockSupport.anyHandler())).thenAnswer(invocation -> { delegatedMessageRef.set(invocation.getArgument(0)); final Handler<ProtonDelivery> dispositionHandler = invocation.getArgument(1); dispositionHandler.handle(sendMsgDeliveryUpdate); return mock(ProtonDelivery.class); }); final AtomicReference<CommandContext> localHandlerCmdContextRef = new AtomicReference<>(); adapterInstanceCommandHandler.putDeviceSpecificCommandHandler(Constants.DEFAULT_TENANT, gatewayId, null, localHandlerCmdContextRef::set); final Message message = getValidCommandMessage(deviceId); final ProtonDelivery delivery = mock(ProtonDelivery.class); mappingAndDelegatingCommandHandler.mapAndDelegateIncomingCommandMessage(Constants.DEFAULT_TENANT, delivery, message); verify(delivery).disposition(any(Accepted.class), eq(true)); assertThat(localHandlerCmdContextRef.get()).isNull(); final Message delegatedMessage = delegatedMessageRef.get(); assertThat(delegatedMessage).isNotNull(); assertThat(delegatedMessage.getAddress()).isEqualTo(String.format("%s/%s/%s", CommandConstants.COMMAND_ENDPOINT, Constants.DEFAULT_TENANT, deviceId)); final String viaProperty = MessageHelper.getApplicationProperty(delegatedMessage.getApplicationProperties(), MessageHelper.APP_PROPERTY_CMD_VIA, String.class); assertThat(viaProperty).isEqualTo(gatewayId); }