AmqpClientActor extends BaseClientActor implements ExceptionListener { static Props propsForTests(final Connection connection, @Nullable final ActorRef proxyActor, final ActorRef connectionActor, final JmsConnectionFactory jmsConnectionFactory) { return Props.create(AmqpClientActor.class, validateConnection(connection), jmsConnectionFactory, proxyActor, connectionActor); } @SuppressWarnings("unused") private  AmqpClientActor(final Connection connection,
            final JmsConnectionFactory jmsConnectionFactory,
            @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor); @SuppressWarnings("unused") private  AmqpClientActor(final Connection connection, @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor); static Props props(final Connection connection, @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor); @Override void postStop(); @Override void onException(final JMSException exception);  }
@Test public void invalidSpecificOptionsThrowConnectionConfigurationInvalidException() { final Map<String, String> specificOptions = new HashMap<>(); specificOptions.put("failover.unknown.option", "100"); specificOptions.put("failover.nested.amqp.vhost", "ditto"); final Connection connection = ConnectivityModelFactory.newConnectionBuilder(createRandomConnectionId(), ConnectionType.AMQP_10, ConnectivityStatus.OPEN, TestConstants.getUriOfNewMockServer()) .specificConfig(specificOptions) .sources(singletonList( ConnectivityModelFactory.newSourceBuilder() .authorizationContext(Authorization.AUTHORIZATION_CONTEXT) .address("source1") .build())) .build(); final ThrowableAssert.ThrowingCallable props1 = () -> AmqpClientActor.propsForTests(connection, ActorRef.noSender(), null, null); final ThrowableAssert.ThrowingCallable props2 = () -> AmqpClientActor.propsForTests(connection, ActorRef.noSender(), null, jmsConnectionFactory); Stream.of(props1, props2).forEach(throwingCallable -> assertThatExceptionOfType(ConnectionConfigurationInvalidException.class) .isThrownBy(throwingCallable) .withMessageContaining("unknown.option")); }
@Test public void testExceptionDuringJMSConnectionCreation() { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (theConnection, exceptionListener) -> { throw JMS_EXCEPTION; }); final ActorRef connectionActor = actorSystem.actorOf(props); connectionActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(new Status.Failure(SESSION_EXCEPTION)); }}; }
@Test public void testConnectionHandling() { new TestKit(actorSystem) {{ final TestProbe aggregator = new TestProbe(actorSystem); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (connection1, exceptionListener) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); watch(amqpClientActor); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); amqpClientActor.tell(RetrieveConnectionStatus.of(CONNECTION_ID, DittoHeaders.empty()), aggregator.ref()); aggregator.expectMsgClass(ResourceStatus.class); amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); amqpClientActor.tell(RetrieveConnectionStatus.of(CONNECTION_ID, DittoHeaders.empty()), aggregator.ref()); aggregator.expectMsgClass(ResourceStatus.class); }}; }
@Test public void testReconnect() { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (connection1, exceptionListener) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); watch(amqpClientActor); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); for (int i = 0; i < 10; ++i) { amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); } amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); }}; }
@Test public void testReconnectAndVerifyConnectionStatus() { new TestKit(actorSystem) {{ final TestProbe aggregator = new TestProbe(actorSystem); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (connection1, exceptionListener) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); watch(amqpClientActor); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); amqpClientActor.tell(RetrieveConnectionStatus.of(CONNECTION_ID, DittoHeaders.empty()), aggregator.ref()); aggregator.expectMsgClass(ResourceStatus.class); final JmsConnectionListener connectionListener = checkNotNull(listenerArgumentCaptor.getValue()); connectionListener.onConnectionInterrupted(DUMMY); verifyConnectionStatus(amqpClientActor, aggregator, ConnectivityStatus.FAILED); connectionListener.onConnectionRestored(DUMMY); verifyConnectionStatus(amqpClientActor, aggregator, ConnectivityStatus.OPEN); amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); }}; }
@Test public void sendCommandDuringInit() { new TestKit(actorSystem) {{ final CountDownLatch latch = new CountDownLatch(1); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> waitForLatchAndReturn(latch, mockConnection)); final ActorRef amqpClientActor = actorSystem.actorOf(props); watch(amqpClientActor); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); latch.countDown(); expectMsg(CONNECTED_SUCCESS); }}; }
@Test public void sendConnectCommandWhenAlreadyConnected() throws JMSException { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); Mockito.verify(mockConnection, Mockito.times(1)).start(); }}; }
@Test public void sendDisconnectWhenAlreadyDisconnected() { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); Mockito.verifyZeroInteractions(mockConnection); }}; }
@Test public void testStartConnectionFails() throws JMSException { new TestKit(actorSystem) {{ doThrow(JMS_EXCEPTION).when(mockConnection).start(); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(new Status.Failure(SESSION_EXCEPTION)); }}; }
@Test public void testCreateSessionFails() throws JMSException { new TestKit(actorSystem) {{ when(mockConnection.createSession(anyInt())).thenThrow(JMS_EXCEPTION); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(new Status.Failure(SESSION_EXCEPTION)); }}; }
@Test public void testCreateConsumerFails() throws JMSException { new TestKit(actorSystem) {{ doReturn(mockSession).when(mockConnection).createSession(anyInt()); when(mockSession.createConsumer(any())).thenThrow(JMS_EXCEPTION); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsgClass(Status.Failure.class); }}; }
@Test public void testCloseConnectionFails() throws JMSException { new TestKit(actorSystem) {{ doThrow(JMS_EXCEPTION).when(mockConnection).close(); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); amqpClientActor.tell(CloseConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); }}; }
@Test public void testConnectionRestoredExpectRecreateSession() throws JMSException { final Target target = TestConstants.Targets.TWIN_TARGET; final JmsMessageConsumer recoveredConsumer = Mockito.mock(JmsMessageConsumer.class); final JmsSession newSession = Mockito.mock(JmsSession.class, withSettings().name("recoveredSession")); when(mockSession.isClosed()).thenReturn(true); doReturn(mockSession) .doReturn(newSession) .when(mockConnection) .createSession(anyInt()); prepareSession(newSession, recoveredConsumer); new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); final ArgumentCaptor<JmsConnectionListener> captor = ArgumentCaptor.forClass(JmsConnectionListener.class); verify(mockConnection).addConnectionListener(captor.capture()); final JmsConnectionListener jmsConnectionListener = captor.getValue(); sendThingEventAndExpectPublish(amqpClientActor, target, () -> getProducerForAddress(target.getAddress())); consumeMockMessage(mockConsumer); expectMsgClass(Command.class); jmsConnectionListener.onConnectionRestored(URI.create("amqp: verify(mockConnection, timeout(2000).times(2)).createSession(anyInt()); verify(mockSession, timeout(2000).times(2)).close(); sendThingEventAndExpectPublish(amqpClientActor, target, () -> getProducerForAddress(target.getAddress())); consumeMockMessage(recoveredConsumer); expectMsgClass(Command.class); }}; }
@Test public void testTargetAddressPlaceholderReplacement() throws JMSException { final Connection connection = TestConstants.createConnection(CONNECTION_ID, TestConstants.Targets.TARGET_WITH_PLACEHOLDER); final String expectedAddress = "target:" + TestConstants.Things.NAMESPACE + "/" + TestConstants.Things.ID + "@" + TopicPath.Channel.TWIN.getName(); new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); final ThingModifiedEvent thingModifiedEvent = TestConstants.thingModified(Collections.emptyList()); final OutboundSignal outboundSignal = OutboundSignalFactory.newOutboundSignal(thingModifiedEvent, singletonList(ConnectivityModelFactory.newTargetBuilder() .address(TestConstants.Targets.TARGET_WITH_PLACEHOLDER.getAddress()) .authorizationContext(Authorization.AUTHORIZATION_CONTEXT) .topics(Topic.TWIN_EVENTS) .build())); amqpClientActor.tell(outboundSignal, getRef()); final ArgumentCaptor<JmsMessage> messageCaptor = ArgumentCaptor.forClass(JmsMessage.class); final MessageProducer messageProducer = getProducerForAddress(expectedAddress); verify(messageProducer, timeout(2000)).send(messageCaptor.capture(), any(CompletionListener.class)); final Message message = messageCaptor.getValue(); assertThat(message).isNotNull(); }}; }
@Test public void testReceiveThingEventAndExpectForwardToJMSProducer() throws JMSException { final Target target = TestConstants.Targets.TWIN_TARGET; new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); sendThingEventAndExpectPublish(amqpClientActor, target, () -> getProducerForAddress(target.getAddress())); }}; }
@Test public void testConsumerRecreationFailureWhenConnected() throws JMSException { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(singleConsumerConnection(), getRef(), getRef(), (ac, el) -> mockConnection); final TestActorRef<AmqpClientActor> amqpClientActorRef = TestActorRef.apply(props, actorSystem); final AmqpClientActor amqpClientActor = amqpClientActorRef.underlyingActor(); amqpClientActorRef.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); final Session mockSession2 = Mockito.mock(Session.class); final MessageConsumer mockConsumer2 = Mockito.mock(JmsMessageConsumer.class); when(mockSession.createConsumer(any())) .thenThrow(new IllegalStateException("expected exception")); doReturn(mockSession2).when(mockConnection).createSession(anyInt()); doReturn(mockConsumer2).when(mockSession2).createConsumer(any()); final ActorRef amqpConsumerActor = amqpClientActor.context().children().toStream() .find(child -> child.path().name().startsWith(AmqpConsumerActor.ACTOR_NAME_PREFIX)) .get(); final Throwable error = new IllegalStateException("Forcibly detached"); final Status.Failure failure = new Status.Failure(new AskTimeoutException("Consumer creation timeout")); amqpClientActor.connectionListener.onConsumerClosed(mockConsumer, error); verify(mockSession, atLeastOnce()).createConsumer(any()); amqpConsumerActor.tell(failure, amqpConsumerActor); verify(mockConnection).createSession(anyInt()); final ArgumentCaptor<MessageListener> captor = ArgumentCaptor.forClass(MessageListener.class); verify(mockConsumer2, timeout(1000).atLeastOnce()).setMessageListener(captor.capture()); final MessageListener messageListener = captor.getValue(); messageListener.onMessage(mockMessage()); expectMsgClass(Command.class); }}; }
@Test public void testTestConnection() { new TestKit(actorSystem) {{ final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(TestConnection.of(connection, DittoHeaders.empty()), getRef()); expectMsgClass(Status.Success.class); }}; }
@Test public void testTestConnectionFailsOnTimeout() throws JMSException { final CountDownLatch latch = new CountDownLatch(1); try { new TestKit(actorSystem) {{ when(mockConnection.createSession(anyInt())) .thenAnswer(invocationOnMock -> waitForLatchAndReturn(latch, mockSession)); final Props props = AmqpClientActor.propsForTests(connection, getRef(), getRef(), (ac, el) -> mockConnection); final ActorRef amqpClientActor = actorSystem.actorOf(props); amqpClientActor.tell(TestConnection.of(connection, DittoHeaders.empty()), getRef()); amqpClientActor.tell(FSM.StateTimeout$.MODULE$, amqpClientActor); expectMsgClass(Status.Failure.class); }}; } finally { latch.countDown(); } }