AmqpConsumerActor extends BaseConsumerActor implements MessageListener, MessageRateLimiterBehavior<String> { static Props props(final ConnectionId connectionId, final ConsumerData consumerData, final ActorRef messageMappingProcessor, final ActorRef jmsActor) { return Props.create(AmqpConsumerActor.class, connectionId, consumerData, messageMappingProcessor, jmsActor); } @SuppressWarnings("unused") private  AmqpConsumerActor(final ConnectionId connectionId, final ConsumerData consumerData,
            final ActorRef messageMappingProcessor, final ActorRef jmsActor); @Override Receive createReceive(); @Override void preStart(); @Override void postStop(); @Override void onMessage(final Message message); @Override void stopMessageConsumerDueToRateLimit(final String reason); @Override void startMessageConsumerDueToRateLimit(); @Override MessageRateLimiter<String> getMessageRateLimiter(); @Override DittoDiagnosticLoggingAdapter log();  }
@Test public void plainStringMappingTest() { new TestKit(actorSystem) {{ final MappingContext mappingContext = ConnectivityModelFactory.newMappingContext( "JavaScript", JavaScriptMessageMapperFactory.createJavaScriptMessageMapperConfigurationBuilder( "plainStringMapping", Collections.emptyMap()) .incomingScript(TestConstants.Mapping.INCOMING_MAPPING_SCRIPT) .outgoingScript(TestConstants.Mapping.OUTGOING_MAPPING_SCRIPT) .build() .getPropertiesAsJson() ); final ActorRef processor = setupActor(getRef(), mappingContext); final Source source = Mockito.mock(Source.class); Mockito.when(source.getAuthorizationContext()) .thenReturn(TestConstants.Authorization.AUTHORIZATION_CONTEXT); Mockito.when(source.getPayloadMapping()).thenReturn(ConnectivityModelFactory.newPayloadMapping("test")); final ActorRef underTest = actorSystem.actorOf(AmqpConsumerActor.props(CONNECTION_ID, consumerData("foo", Mockito.mock(MessageConsumer.class), source), processor, getRef())); final String plainPayload = "hello world!"; final String correlationId = "cor-"; underTest.tell(getJmsMessage(plainPayload, correlationId), null); final Command<?> command = expectMsgClass(Command.class); assertThat(command.getType()).isEqualTo(ModifyAttribute.TYPE); assertThat(command.getDittoHeaders().getCorrelationId()).contains(correlationId); assertThat(((ModifyAttribute) command).getAttributePointer()).isEqualTo(JsonPointer.of("/foo")); assertThat(((ModifyAttribute) command).getAttributeValue()).isEqualTo(JsonValue.of(plainPayload)); }}; }
@Test public void plainStringMappingMultipleTest() { new TestKit(actorSystem) {{ final MappingContext mappingContext = ConnectivityModelFactory.newMappingContext( "JavaScript", JavaScriptMessageMapperFactory.createJavaScriptMessageMapperConfigurationBuilder( "plainStringMultiMapping", Collections.emptyMap()) .incomingScript(TestConstants.Mapping.INCOMING_MAPPING_SCRIPT .replace("return msg;", "return [msg, msg, msg];")) .outgoingScript(TestConstants.Mapping.OUTGOING_MAPPING_SCRIPT) .build() .getPropertiesAsJson() ); final ActorRef processor = setupActor(getRef(), mappingContext); final Source source = Mockito.mock(Source.class); Mockito.when(source.getAuthorizationContext()) .thenReturn(TestConstants.Authorization.AUTHORIZATION_CONTEXT); Mockito.when(source.getPayloadMapping()).thenReturn(ConnectivityModelFactory.newPayloadMapping("test")); final ActorRef underTest = actorSystem.actorOf(AmqpConsumerActor.props(CONNECTION_ID, consumerData("foo", Mockito.mock(MessageConsumer.class), source), processor, getRef())); final String plainPayload = "hello world!"; final String correlationId = "cor-"; underTest.tell(getJmsMessage(plainPayload, correlationId), null); for (int i = 0; i < 3; i++) { final Command<?> command = expectMsgClass(Command.class); assertThat(command.getType()).isEqualTo(ModifyAttribute.TYPE); assertThat(command.getDittoHeaders().getCorrelationId()).contains(correlationId); assertThat(((ModifyAttribute) command).getAttributePointer()).isEqualTo(JsonPointer.of("/foo")); assertThat(((ModifyAttribute) command).getAttributeValue()).isEqualTo(JsonValue.of(plainPayload)); } }}; }
@Test public void jmsMessageWithNullPropertyAndNullContentTypeTest() throws JMSException { new TestKit(actorSystem) {{ final ActorRef testActor = getTestActor(); final ActorRef processor = setupActor(testActor, null); final Source source = Mockito.mock(Source.class); Mockito.when(source.getAuthorizationContext()) .thenReturn(TestConstants.Authorization.AUTHORIZATION_CONTEXT); Mockito.when(source.getHeaderMapping()) .thenReturn(Optional.of(ConnectivityModelFactory.newHeaderMapping( Collections.singletonMap("correlation-id", "{{ header:correlation-id }}") ))); final ActorRef underTest = actorSystem.actorOf( AmqpConsumerActor.props(CONNECTION_ID, consumerData("foo123", Mockito.mock(MessageConsumer.class), source), processor, getRef())); final String correlationId = "cor-"; final String plainPayload = "{ \"topic\": \"org.eclipse.ditto.test/testThing/things/twin/commands/modify\"," + " \"headers\":{\"device_id\":\"org.eclipse.ditto.test:testThing\"}," + " \"path\": \"/features/point/properties/x\", \"value\": 42 }"; final AmqpJmsTextMessageFacade messageFacade = new AmqpJmsTextMessageFacade(); messageFacade.setApplicationProperty("JMSXDeliveryCount", null); messageFacade.setText(plainPayload); messageFacade.setContentType(null); messageFacade.setCorrelationId(correlationId); final JmsMessage jmsMessage = messageFacade.asJmsMessage(); underTest.tell(jmsMessage, null); final Command<?> command = expectMsgClass(Command.class); assertThat(command.getType()).isEqualTo(ModifyFeatureProperty.TYPE); assertThat(command.getDittoHeaders().getCorrelationId()).contains(correlationId); assertThat(command.getDittoHeaders().getContentType()).isEmpty(); assertThat(command.getDittoHeaders().get("JMSXDeliveryCount")).isNull(); assertThat(((ModifyFeatureProperty) command).getPropertyPointer()).isEqualTo(JsonPointer.of("/x")); assertThat(((ModifyFeatureProperty) command).getPropertyValue()).isEqualTo(JsonValue.of(42)); }}; }