ConnectionPersistenceActor extends AbstractShardedPersistenceActor<ConnectivityCommand, Connection, ConnectionId, ConnectionState,
        ConnectivityEvent> { public static Props props(final ConnectionId connectionId, final DittoProtocolSub dittoProtocolSub, final ActorRef proxyActor, final ClientActorPropsFactory propsFactory, @Nullable final ConnectivityCommandInterceptor commandValidator ) { return Props.create(ConnectionPersistenceActor.class, connectionId, dittoProtocolSub, proxyActor, propsFactory, commandValidator, CLIENT_ACTORS_PER_NODE); } ConnectionPersistenceActor(final ConnectionId connectionId,
            final DittoProtocolSub dittoProtocolSub,
            final ActorRef proxyActor,
            final ClientActorPropsFactory propsFactory,
            @Nullable final ConnectivityCommandInterceptor customCommandValidator,
            final int clientActorsPerNode); static Props props(final ConnectionId connectionId,
            final DittoProtocolSub dittoProtocolSub,
            final ActorRef proxyActor,
            final ClientActorPropsFactory propsFactory,
            @Nullable final ConnectivityCommandInterceptor commandValidator
    ); @Override String persistenceId(); @Override String journalPluginId(); @Override String snapshotPluginId(); @Override void postStop(); @Override void onMutation(final Command command, final ConnectivityEvent event, final WithDittoHeaders response,
            final boolean becomeCreated, final boolean becomeDeleted); static final String PERSISTENCE_ID_PREFIX; static final String JOURNAL_PLUGIN_ID; static final String SNAPSHOT_PLUGIN_ID; }
@Test public void manageConnection() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor( connectionId, actorSystem, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref()), TestConstants.dummyDittoProtocolSub(pubSubMediator, dittoProtocolSubMock), pubSubMediator ); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); expectRemoveSubscriber(1); expectSubscribe(TWIN_AND_LIVE_EVENTS, SUBJECTS); underTest.tell(closeConnection, getRef()); probe.expectMsg(closeConnection); expectMsg(closeConnectionResponse); expectRemoveSubscriber(2); underTest.tell(deleteConnection, getRef()); expectMsg(deleteConnectionResponse); probe.expectNoMessage(); expectTerminated(underTest); }}; }
@Test public void deleteConnectionUpdatesSubscriptions() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor( connectionId, actorSystem, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref()), TestConstants.dummyDittoProtocolSub(pubSubMediator, dittoProtocolSubMock), pubSubMediator ); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); expectRemoveSubscriber(1); expectSubscribe(TWIN_AND_LIVE_EVENTS, SUBJECTS); underTest.tell(deleteConnection, getRef()); expectMsg(deleteConnectionResponse); expectRemoveSubscriber(2); probe.expectNoMessage(); expectTerminated(underTest); }}; }
@Test public void manageConnectionWith2Clients() throws Exception { startSecondActorSystemAndJoinCluster(); new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createClosedConnectionWith2Clients, getRef()); expectMsgClass(CreateConnectionResponse.class); underTest.tell(openConnection, getRef()); probe.expectMsg(openConnection); expectMsgClass(OpenConnectionResponse.class); underTest.tell(thingModified, getRef()); final Object outboundSignal = probe.expectMsgClass(Object.class); assertThat(outboundSignal).isInstanceOf(OutboundSignal.class); underTest.tell(closeConnection, getRef()); probe.expectMsg(closeConnection); expectMsg(closeConnectionResponse); underTest.tell(deleteConnection, getRef()); expectMsg(deleteConnectionResponse); expectTerminated(underTest); }}; }
@Test public void createConnectionAfterDeleted() { new TestKit(actorSystem) {{ final TestProbe clientActorMock = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(clientActorMock.ref())); watch(underTest); underTest.tell(createConnection, getRef()); clientActorMock.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(deleteConnection, getRef()); expectMsg(deleteConnectionResponse); underTest.tell(createConnection, getRef()); expectMsg(createConnectionResponse); clientActorMock.expectMsg(openConnection); }}; }
@Test public void openConnectionAfterDeletedFails() { new TestKit(actorSystem) {{ final TestProbe clientActorMock = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(clientActorMock.ref())); watch(underTest); underTest.tell(createConnection, getRef()); clientActorMock.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(deleteConnection, getRef()); expectMsg(deleteConnectionResponse); underTest.tell(openConnection, getRef()); expectMsg(connectionNotAccessibleException); }}; }
@Test public void createConnectionInClosedState() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createClosedConnection, getRef()); expectMsg(createClosedConnectionResponse); probe.expectNoMessage(); }}; }
@Test public void modifyConnectionInClosedState() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(closeConnection, getRef()); probe.expectMsg(closeConnection); expectMsg(closeConnectionResponse); underTest.tell(modifyClosedConnection, getRef()); probe.expectNoMessage(); expectMsg(modifyConnectionResponse); }}; }
@Test public void retrieveMetricsInClosedStateDoesNotStartClientActor() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createClosedConnection, getRef()); expectMsg(createClosedConnectionResponse); probe.expectNoMessage(); underTest.tell(RetrieveConnectionMetrics.of(connectionId, DittoHeaders.empty()), getRef()); probe.expectNoMessage(); final RetrieveConnectionMetricsResponse metricsResponse = RetrieveConnectionMetricsResponse.getBuilder(connectionId, DittoHeaders.empty()) .connectionMetrics(ConnectivityModelFactory.emptyConnectionMetrics()) .sourceMetrics(ConnectivityModelFactory.emptySourceMetrics()) .targetMetrics(ConnectivityModelFactory.emptyTargetMetrics()) .build(); expectMsg(metricsResponse); }}; }
@Test public void modifyConnectionClosesAndRestartsClientActor() { new TestKit(actorSystem) {{ final TestProbe mockClientProbe = TestProbe.apply(actorSystem); final TestProbe commandSender = TestProbe.apply(actorSystem); final AtomicReference<Connection> latestConnection = new AtomicReference<>(); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, proxyActor, (connection, proxyActor, connectionActor) -> { latestConnection.set(connection); return MockClientActor.props(mockClientProbe.ref()); }, TestConstants.dummyDittoProtocolSub(pubSubMediator, dittoProtocolSubMock), pubSubMediator); underTest.tell(createConnection, commandSender.ref()); mockClientProbe.expectMsg(openConnection); commandSender.expectMsg(createConnectionResponse); final ActorRef clientActor = watch(mockClientProbe.sender()); expectSubscribe(TWIN_AND_LIVE_EVENTS, SUBJECTS); underTest.tell(modifyConnection, commandSender.ref()); mockClientProbe.expectMsg(CloseConnection.of(connectionId, DittoHeaders.empty())); expectRemoveSubscriber(1); expectTerminated(clientActor); mockClientProbe.expectMsg(openConnection); commandSender.expectMsg(modifyConnectionResponse); expectSubscribe(TWIN_AND_LIVE_EVENTS_AND_MESSAGES, SUBJECTS); Awaitility.await().untilAtomic(latestConnection, CoreMatchers.is(modifyConnection.getConnection())); }}; }
@Test public void recoverOpenConnection() { new TestKit(actorSystem) {{ final TestProbe mockClientProbe = TestProbe.apply(actorSystem); ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(mockClientProbe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); expectMsg(createConnectionResponse); mockClientProbe.expectMsg(OpenConnection.of(connectionId, DittoHeaders.empty())); getSystem().stop(underTest); expectTerminated(underTest); final TestProbe recoveredMockClientProbe = TestProbe.apply(actorSystem); underTest = Retry.untilSuccess( () -> TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props( recoveredMockClientProbe.ref()))); recoveredMockClientProbe.expectMsg(OpenConnection.of(connectionId, DittoHeaders.empty())); final ActorRef recoveredActor = underTest; Awaitility.await().untilAsserted(() -> { recoveredActor.tell(retrieveConnectionStatus, getRef()); expectMsg(retrieveConnectionStatusOpenResponse); }); }}; }
@Test public void exceptionDuringClientActorPropsCreation() { new TestKit(actorSystem) {{ final Props connectionActorProps = ConnectionPersistenceActor.props(TestConstants.createRandomConnectionId(), TestConstants.dummyDittoProtocolSub(pubSubMediator), proxyActor, (connection, proxyActor, connectionActor) -> { throw ConnectionConfigurationInvalidException.newBuilder("validation failed...") .build(); }, null); final TestKit parent = new TestKit(actorSystem); final ActorRef connectionActorRef = watch(parent.childActorOf(connectionActorProps)); connectionActorRef.tell(createConnection, parent.getRef()); final Exception exception = parent.expectMsgClass(ConnectionConfigurationInvalidException.class); assertThat(exception).hasMessageContaining("validation failed..."); }}; }
@Test public void exceptionDueToCustomValidator() { new TestKit(actorSystem) {{ final Props connectionActorProps = ConnectionPersistenceActor.props(TestConstants.createRandomConnectionId(), TestConstants.dummyDittoProtocolSub(pubSubMediator), proxyActor, mockClientActorPropsFactory, (command, connection) -> { throw ConnectionUnavailableException.newBuilder(connectionId) .dittoHeaders(command.getDittoHeaders()) .message("not valid") .build(); }); final TestKit parent = new TestKit(actorSystem); final ActorRef connectionActorRef = watch(parent.childActorOf(connectionActorProps)); connectionActorRef.tell(createConnection, parent.getRef()); final ConnectionUnavailableException exception = parent.expectMsgClass(ConnectionUnavailableException.class); assertThat(exception).hasMessageContaining("not valid"); }}; }
@Test public void testResetConnectionMetrics() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(resetConnectionMetrics, getRef()); probe.expectMsg(resetConnectionMetrics); final ResetConnectionMetricsResponse resetResponse = ResetConnectionMetricsResponse.of(connectionId, DittoHeaders.empty()); expectMsg(resetResponse); }}; }
@Test public void testConnectionActorRespondsToCleanupCommand() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell( CleanupPersistence.of(DefaultEntityId.of( ConnectionPersistenceActor.PERSISTENCE_ID_PREFIX + connectionId), DittoHeaders.empty()), getRef()); expectMsg(CleanupPersistenceResponse.success( DefaultEntityId.of(ConnectionPersistenceActor.PERSISTENCE_ID_PREFIX + connectionId), DittoHeaders.empty())); }}; }
@Test public void enableConnectionLogs() { final EnableConnectionLogs enableConnectionLogs = EnableConnectionLogs.of(connectionId, DittoHeaders.empty()); final EnableConnectionLogsResponse enableConnectionLogsResponse = EnableConnectionLogsResponse.of(connectionId, enableConnectionLogs.getDittoHeaders()); new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(enableConnectionLogs, getRef()); probe.expectMsg(enableConnectionLogs); expectMsg(enableConnectionLogsResponse); }}; }
@Test public void retrieveLogsInClosedStateDoesNotStartClientActor() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createClosedConnection, getRef()); expectMsg(createClosedConnectionResponse); probe.expectNoMessage(); underTest.tell(RetrieveConnectionLogs.of(connectionId, DittoHeaders.empty()), getRef()); probe.expectNoMessage(); final RetrieveConnectionLogsResponse logsResponse = RetrieveConnectionLogsResponse.of(connectionId, Collections.emptyList(), null, null, DittoHeaders.empty()); expectMsg(logsResponse); }}; }
@Test public void retrieveLogsIsAggregated() { final Instant now = Instant.now(); final RetrieveConnectionLogsResponse innerResponse = RetrieveConnectionLogsResponse.of(connectionId, TestConstants.Monitoring.LOG_ENTRIES, now.minusSeconds(312), now.plusSeconds(123), DittoHeaders.empty()); new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); final RetrieveConnectionLogs retrieveConnectionLogs = RetrieveConnectionLogs.of(connectionId, DittoHeaders.empty()); underTest.tell(retrieveConnectionLogs, getRef()); probe.expectMsg(retrieveConnectionLogs); final ActorRef aggregatorActor = probe.sender(); probe.send(aggregatorActor, innerResponse); expectMsg(innerResponse); }}; }
@Test public void resetConnectionLogs() { final ResetConnectionLogs resetConnectionLogs = ResetConnectionLogs.of(connectionId, DittoHeaders.empty()); final ResetConnectionLogsResponse expectedResponse = ResetConnectionLogsResponse.of(connectionId, resetConnectionLogs.getDittoHeaders()); new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(resetConnectionLogs, getRef()); probe.expectMsg(resetConnectionLogs); expectMsg(expectedResponse); }}; }
@Test public void enabledConnectionLogsAreEnabledAgainAfterModify() { final EnableConnectionLogs enableConnectionLogs = EnableConnectionLogs.of(connectionId, DittoHeaders.empty()); final EnableConnectionLogsResponse enableConnectionLogsResponse = EnableConnectionLogsResponse.of(connectionId, enableConnectionLogs.getDittoHeaders()); new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref()), TestConstants.dummyDittoProtocolSub(pubSubMediator, dittoProtocolSubMock), pubSubMediator); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); expectAnySubscribe(); underTest.tell(enableConnectionLogs, getRef()); probe.expectMsg(enableConnectionLogs); expectMsg(enableConnectionLogsResponse); underTest.tell(modifyConnection, getRef()); probe.expectMsg(closeConnection); probe.expectMsg(openConnection); expectMsg(modifyConnectionResponse); probe.expectMsg(enableConnectionLogs); }}; }
@Test public void disabledConnectionLogsAreNotEnabledAfterModify() { new TestKit(actorSystem) {{ final TestProbe probe = TestProbe.apply(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, pubSubMediator, proxyActor, (connection, concierge, connectionActor) -> MockClientActor.props(probe.ref())); watch(underTest); underTest.tell(createConnection, getRef()); probe.expectMsg(openConnection); expectMsg(createConnectionResponse); underTest.tell(modifyConnection, getRef()); probe.expectMsg(closeConnection); probe.expectMsg(openConnection); expectMsg(modifyConnectionResponse); probe.expectNoMsg(); }}; }
@Test public void testHandleSignalWithAcknowledgementRequest() { new TestKit(actorSystem) {{ final TestKit probe = new TestKit(actorSystem); final ActorRef underTest = TestConstants.createConnectionSupervisorActor(connectionId, actorSystem, proxyActor, (connection, connectionActor, proxyActor) -> TestActor.props(probe), TestConstants.dummyDittoProtocolSub(pubSubMediator, dittoProtocolSubMock), pubSubMediator); watch(underTest); underTest.tell(createConnection, getRef()); expectMsgClass(CreateConnectionResponse.class); expectAnySubscribe(); final AcknowledgementLabel acknowledgementLabel = AcknowledgementLabel.of("test-ack"); final DittoHeaders dittoHeaders = DittoHeaders.newBuilder() .acknowledgementRequest(AcknowledgementRequest.of(acknowledgementLabel)) .readGrantedSubjects(Collections.singleton(TestConstants.Authorization.SUBJECT)) .timeout("2s") .randomCorrelationId() .build(); final AttributeModified attributeModified = AttributeModified.of( TestConstants.Things.THING_ID, JsonPointer.of("hello"), JsonValue.of("world!"), 5L, dittoHeaders); underTest.tell(attributeModified, getRef()); final OutboundSignal unmappedOutboundSignal = probe.expectMsgClass(OutboundSignal.class); assertThat(unmappedOutboundSignal.getSource()).isEqualTo(attributeModified); final Acknowledgement acknowledgement = Acknowledgement.of(acknowledgementLabel, TestConstants.Things.THING_ID, HttpStatusCode.OK, dittoHeaders); underTest.tell(acknowledgement, getRef()); expectMsg(acknowledgement); }}; }