RetrieveConnectionLogsAggregatorActor extends AbstractActor { public static Props props(final Connection connection, final ActorRef sender, final DittoHeaders originalHeaders, final Duration timeout, final long maxLogSizeBytes) { return Props.create(RetrieveConnectionLogsAggregatorActor.class, connection, sender, originalHeaders, timeout, maxLogSizeBytes); } @SuppressWarnings("unused") private  RetrieveConnectionLogsAggregatorActor(final Connection connection, final ActorRef sender,
            final DittoHeaders originalHeaders, final Duration timeout, final long maxLogSizeBytes); static Props props(final Connection connection, final ActorRef sender,
            final DittoHeaders originalHeaders, final Duration timeout, final long maxLogSizeBytes); @Override Receive createReceive(); @Override void preStart();  }
@Test public void withOneClient() { new TestKit(actorSystem) {{ final TestProbe sender = TestProbe.apply(actorSystem); final Connection connection = createConnectionWithClients(1); final RetrieveConnectionLogsResponse expectedResponse = createRetrieveConnectionLogsResponse(connection.getId(), Instant.now().minusSeconds(123), Instant.now().plusSeconds(444)); final ActorRef underTest = childActorOf( RetrieveConnectionLogsAggregatorActor.props(connection, sender.ref(), DITTO_HEADERS, DEFAULT_TIMEOUT, LOGGER_CONFIG.maxLogSizeInBytes())); underTest.tell(expectedResponse, getRef()); sender.expectMsg(expectedResponse); }}; }
@Test public void withMultipleClients() { new TestKit(actorSystem) {{ final TestProbe sender = TestProbe.apply(actorSystem); final Connection connection = createConnectionWithClients(3); final Instant since = Instant.now().minusSeconds(333); final Instant until = Instant.now().plusSeconds(555); final Collection<RetrieveConnectionLogsResponse> responses = Arrays.asList( createRetrieveConnectionLogsResponse(connection.getId(), since, until), createRetrieveConnectionLogsResponse(connection.getId(), since, until), createRetrieveConnectionLogsResponse(connection.getId(), since, until) ); final RetrieveConnectionLogsResponse expectedResponse = createExpectedResponse(responses); final ActorRef underTest = childActorOf( RetrieveConnectionLogsAggregatorActor.props(connection, sender.ref(), DITTO_HEADERS, DEFAULT_TIMEOUT, LOGGER_CONFIG.maxLogSizeInBytes())); responses.forEach(response -> underTest.tell(response, getRef())); sender.expectMsg(expectedResponse); }}; }
@Test public void withMultipleClientsRespectsMaxLogSize() throws InterruptedException { new TestKit(actorSystem) {{ final TestProbe sender = TestProbe.apply(actorSystem); final Connection connection = createConnectionWithClients(3); final Instant since = Instant.now().minusSeconds(333); final Instant until = Instant.now().plusSeconds(555); final Collection<RetrieveConnectionLogsResponse> responses = Arrays.asList( createRetrieveConnectionLogsResponse(connection.getId(), since, until, maxSizeLogEntries()), createRetrieveConnectionLogsResponse(connection.getId(), since, until, maxSizeLogEntries()), createRetrieveConnectionLogsResponse(connection.getId(), since, until, maxSizeLogEntries()) ); final ActorRef underTest = childActorOf( RetrieveConnectionLogsAggregatorActor.props(connection, sender.ref(), DITTO_HEADERS, DEFAULT_TIMEOUT, LOGGER_CONFIG.maxLogSizeInBytes())); responses.forEach(response -> underTest.tell(response, getRef())); final RetrieveConnectionLogsResponse retrieveConnectionLogsResponse = sender.expectMsgClass(RetrieveConnectionLogsResponse.class); assertThat((Long) retrieveConnectionLogsResponse.getConnectionLogs().stream() .map(LogEntry::toJsonString) .map(String::length) .mapToLong(Integer::intValue) .sum()) .isLessThan(LOGGER_CONFIG.maxLogSizeInBytes()); }}; }
@Test public void withTimeoutAndWithoutMessages() { new TestKit(actorSystem) {{ final TestProbe sender = TestProbe.apply(actorSystem); final Connection connection = createConnectionWithClients(1); final Duration shortTimeout = Duration.ofMillis(10); final ActorRef underTest = childActorOf( RetrieveConnectionLogsAggregatorActor.props(connection, sender.ref(), DITTO_HEADERS, shortTimeout, LOGGER_CONFIG.maxLogSizeInBytes())); sender.expectMsgClass(ConnectionTimeoutException.class); }}; }
@Test public void withTimeoutWithMessages() { new TestKit(actorSystem) {{ final TestProbe sender = TestProbe.apply(actorSystem); final Duration aggregatorTimeout = Duration.ofSeconds(5); final FiniteDuration expectMessageTimeout = FiniteDuration.apply(aggregatorTimeout.getSeconds() + 1, TimeUnit.SECONDS); final Connection connection = createConnectionWithClients(2); final ActorRef underTest = childActorOf( RetrieveConnectionLogsAggregatorActor.props(connection, sender.ref(), DITTO_HEADERS, aggregatorTimeout, LOGGER_CONFIG.maxLogSizeInBytes())); final RetrieveConnectionLogsResponse expectedResponse = createRetrieveConnectionLogsResponse(connection.getId(), Instant.now().minusSeconds(123), Instant.now().plusSeconds(444)); underTest.tell(expectedResponse, getRef()); sender.expectMsg(expectMessageTimeout, expectedResponse); }}; }