diff --git a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java index 5f21ef2b..071f79ef 100644 --- a/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java +++ b/launchdarkly-android-client-sdk/src/test/java/com/launchdarkly/sdk/android/DirectEventProcessorTest.java @@ -1227,6 +1227,68 @@ public ScheduledFuture scheduleWithFixedDelay(Runnable command, long initialD } } + @Test + public void neitherGoingOfflineNorComingBackOnlineDeliversOnItsOwn() throws Exception { + try (HttpServer server = startEventsServer()) { + EventProcessor eventProcessor = makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + eventProcessor.setOffline(true); + eventProcessor.setOffline(true); + eventProcessor.setOffline(false); + eventProcessor.setOffline(false); + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + assertEquals(1, countEventsOfKind(collectDelivered(server), "custom")); + } finally { + eventProcessor.close(); + } + } + } + + @Test + public void thePeriodicFlushPostsNothingWhenThereIsNothingToSend() throws Exception { + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), null, 50, 60_000, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler); + try { + eventProcessor.setOffline(false); + + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + + @Test + public void eachPeriodicFlushDeliversWhatWasRecordedSinceTheLastOne() throws Exception { + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), null, 50, 60_000, DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, + scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordCustomEvent(CONTEXT, "first", LDValue.ofNull(), null); + assertTrue(server.getRecorder().requireRequest(5, TimeUnit.SECONDS).getBody() + .contains("\"key\":\"first\"")); + + eventProcessor.recordCustomEvent(CONTEXT, "second", LDValue.ofNull(), null); + String later = server.getRecorder().requireRequest(5, TimeUnit.SECONDS).getBody(); + assertTrue(later.contains("\"key\":\"second\"")); + assertFalse(later.contains("\"key\":\"first\"")); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void eventsOnTheirWayIntoTheStoreStillCountAgainstCapacity() throws Exception { // A commit takes the pending events and encodes them before staging, and for that long they @@ -1302,6 +1364,74 @@ boolean stageReserved(byte[] serializedEvent) { } } + @Test + public void aDeliveryNeverClosesABatchPartwayThroughAnotherCommitsRun() throws Exception { + // Events committed together go out together. A batch closed while a commit has staged part of + // its run would leave the rest for a later delivery. + CountDownLatch firstStaged = new CountDownLatch(1); + CountDownLatch releaseCommit = new CountDownLatch(1); + AtomicReference beforeClosing = new AtomicReference<>(); + EventStore store = new EventStore(eventsDirectory.newFolder(), DEFAULT_CAPACITY, true, + logging.logger) { + @Override + boolean stageReserved(byte[] serializedEvent) { + boolean result = super.stageReserved(serializedEvent); + if (new String(serializedEvent, StandardCharsets.UTF_8).contains("\"key\":\"first\"")) { + firstStaged.countDown(); + awaitQuietly(releaseCommit, 5, TimeUnit.SECONDS); + } + return result; + } + + @Override + List closeBatch() { + Runnable before = beforeClosing.getAndSet(null); + if (before != null) { + before.run(); + } + try { + return super.closeBatch(); + } finally { + if (before != null) { + releaseCommit.countDown(); + } + } + } + }; + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + try (HttpServer server = startEventsServer()) { + DirectEventProcessor eventProcessor = makeEventProcessor(server, store, scheduler); + Thread[] competing = new Thread[1]; + beforeClosing.set(() -> { + eventProcessor.recordCustomEvent(CONTEXT, "first", LDValue.ofNull(), null); + eventProcessor.recordCustomEvent(CONTEXT, "second", LDValue.ofNull(), null); + competing[0] = new Thread(eventProcessor::flush); + competing[0].start(); + // Long enough for an unguarded commit to stage "first" and stop short of "second". + awaitQuietly(firstStaged, 500, TimeUnit.MILLISECONDS); + }); + try { + eventProcessor.setOffline(false); + awaitFlush(eventProcessor, 10, TimeUnit.SECONDS); + competing[0].join(5_000); + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + + boolean sawFirst = false; + while (server.getRecorder().count() > 0) { + String body = server.getRecorder().requireRequest(1, TimeUnit.SECONDS).getBody(); + assertEquals("a batch held one of two events committed together: " + body, + body.contains("\"key\":\"first\""), body.contains("\"key\":\"second\"")); + sawFirst |= body.contains("\"key\":\"first\""); + } + assertTrue("the events committed together were never delivered", sawFirst); + } finally { + releaseCommit.countDown(); + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void storedEventsPastTheMaximumAgeAreDiscardedAsDroppedWithoutFailingAFlush() throws Exception { assumePersisting(); @@ -2085,6 +2215,36 @@ public void flushWithTimeoutReportsDeliveredEvents() throws Exception { } } + @Test + public void aDeliveryRecordsHowManyEventsItsBatchCarriedForTheDiagnosticEvent() throws Exception { + DiagnosticStore diagnosticStore = makeDiagnosticStore(); + try (HttpServer server = startEventsServer()) { + ScheduledExecutorService scheduler = EventUtil.makeEventsTaskExecutor(); + DirectEventProcessor eventProcessor = makeEventProcessor(new StubEventSender(), + server.getUri(), diagnosticStore, NO_PERIODIC_FLUSH_MILLIS, 60_000, + DirectEventProcessor.DEFAULT_CLOSE_BUDGET_MILLIS, scheduler); + try { + eventProcessor.setOffline(false); + eventProcessor.recordIdentifyEvent(CONTEXT); + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + recordEvaluation(eventProcessor, false, null); + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + + // The identify, the custom event and the summary. + assertEquals(3, diagnosticStore.createEventAndReset(0, 0).getJsonValue() + .get("eventsInLastBatch").intValue()); + + assertTrue(awaitFlush(eventProcessor, 10, TimeUnit.SECONDS)); + assertEquals("a flush with nothing to send recorded a batch", 0, + diagnosticStore.createEventAndReset(0, 0).getJsonValue() + .get("eventsInLastBatch").intValue()); + } finally { + eventProcessor.close(); + scheduler.shutdownNow(); + } + } + } + @Test public void aCommitNeverSplitsAnEvaluationAcrossTwoCommits() throws Exception { // The other half of the atomicity invariant. close() only ever commits once, so it can show @@ -3411,6 +3571,23 @@ public void aFlushBypassingTheBackoffAskedForBehindAnOrdinaryOneStillPosts() thr } } + @Test + public void aFlushBypassingTheBackoffPostsNothingWhileOffline() throws Exception { + try (HttpServer server = startEventsServer()) { + DirectEventProcessor eventProcessor = + (DirectEventProcessor) makeEventProcessor(server, DEFAULT_CAPACITY); + try { + eventProcessor.recordCustomEvent(CONTEXT, "an-event", LDValue.ofNull(), null); + eventProcessor.setOffline(true); + + assertFalse(eventProcessor.flushBypassingBackoff().get(10, TimeUnit.SECONDS)); + server.getRecorder().requireNoRequests(300, TimeUnit.MILLISECONDS); + } finally { + eventProcessor.close(); + } + } + } + @Test public void aBatchIsDroppedOnlyForTwo400sInARow() throws Exception { try (HttpServer server = HttpServer.start(Handlers.sequential(Handlers.status(400), @@ -4794,8 +4971,8 @@ private DirectEventProcessor makeEventProcessor(EventSender diagnosticSender, return makeEventProcessor(diagnosticSender, eventsUri, diagnosticStore, flushIntervalMillis, diagnosticIntervalMillis, closeBudgetMillis, scheduler, diagnosticExecutor, EventStore.create(eventsDirectory.getRoot(), MOBILE_KEY, - DEFAULT_CAPACITY, true, logging.logger), - DEFAULT_CAPACITY); + DEFAULT_CAPACITY, persistence() != EventPersistence.DISABLED, logging.logger), + DEFAULT_CAPACITY, persistence() == EventPersistence.IMMEDIATE); } private DirectEventProcessor makeEventProcessor(EventStore store, int capacity,