RabbitMQClientActor extends BaseClientActor { static Props propsForTests(final Connection connection, @Nullable final ActorRef proxyActor, final ActorRef connectionActor, final RabbitConnectionFactoryFactory rabbitConnectionFactoryFactory) { return Props.create(RabbitMQClientActor.class, validateConnection(connection), rabbitConnectionFactoryFactory, proxyActor, connectionActor); } @SuppressWarnings("unused") private  RabbitMQClientActor(final Connection connection,
            final RabbitConnectionFactoryFactory rabbitConnectionFactoryFactory,
            @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor); @SuppressWarnings("unused") private  RabbitMQClientActor(final Connection connection, @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor); static Props props(final Connection connection, @Nullable final ActorRef proxyActor,
            final ActorRef connectionActor);  }
@Test public void invalidTargetFormatThrowsConnectionConfigurationInvalidException() { final Connection connection = ConnectivityModelFactory.newConnectionBuilder(CONNECTION_ID, ConnectionType.AMQP_091, ConnectivityStatus.OPEN, TestConstants.getUriOfNewMockServer()) .targets(Collections.singletonList(ConnectivityModelFactory.newTargetBuilder() .address("exchangeOnly") .authorizationContext(TestConstants.Authorization.AUTHORIZATION_CONTEXT) .topics(Topic.TWIN_EVENTS) .build())) .build(); final ThrowableAssert.ThrowingCallable props1 = () -> RabbitMQClientActor.propsForTests(connection, Actor.noSender(), Actor.noSender(), null); final ThrowableAssert.ThrowingCallable props2 = () -> RabbitMQClientActor.propsForTests(connection, Actor.noSender(), Actor.noSender(), rabbitConnectionFactoryFactory ); Stream.of(props1, props2) .forEach(throwingCallable -> assertThatExceptionOfType(ConnectionConfigurationInvalidException.class) .isThrownBy(throwingCallable) .withMessageContaining("exchangeOnly") .withNoCause() ); }
@Test public void testExceptionDuringConnectionFactoryCreation() { new TestKit(actorSystem) {{ final Props props = RabbitMQClientActor.propsForTests(connection, getRef(), getRef(), (con, exHandler) -> { throw CUSTOM_EXCEPTION; }) .withDispatcher(CallingThreadDispatcher.Id()); final ActorRef connectionActor = actorSystem.actorOf(props); connectionActor.tell(OpenConnection.of(CONNECTION_ID, DittoHeaders.empty()), getRef()); expectMsgClass(Status.Failure.class); }}; }
@Test public void testConnectionWithoutPublisherHandling() { new TestKit(actorSystem) {{ final ConnectionId randomConnectionId = TestConstants.createRandomConnectionId(); final Connection connectionWithoutTargets = TestConstants.createConnection(randomConnectionId, new Target[0]); final Props props = RabbitMQClientActor.propsForTests(connectionWithoutTargets, getRef(), getRef(), (con, exHandler) -> mockConnectionFactory).withDispatcher(CallingThreadDispatcher.Id()); final ActorRef rabbitClientActor = actorSystem.actorOf(props); watch(rabbitClientActor); rabbitClientActor.tell(OpenConnection.of(randomConnectionId, DittoHeaders.empty()), getRef()); expectMsg(CONNECTED_SUCCESS); rabbitClientActor.tell(CloseConnection.of(randomConnectionId, DittoHeaders.empty()), getRef()); expectMsg(DISCONNECTED_SUCCESS); }}; }