diff --git a/osgi-bundles/bundles/logger/html-sse-demo.md b/osgi-bundles/bundles/logger/html-sse-demo.md new file mode 100644 index 000000000..dac631a8a --- /dev/null +++ b/osgi-bundles/bundles/logger/html-sse-demo.md @@ -0,0 +1,168 @@ +### HTML SSE Demo + +1. Build the project: `mvn clean install -pl osgi-bundles/bundles/logger` +2. Copy the JAR to `platform` plugin directory +3. Copy HTML content below to `` +4. Run this using `python3 -m http.server 3000 -d ` or `npx serve -l 3000 ` + +```html + + + + + SSE Logger Client + + + +

SSE Logger Client

+
+
+ + +
+
+ + +
+
+ + +
+ + + ● Disconnected +
+
+ + + + + +``` \ No newline at end of file diff --git a/osgi-bundles/bundles/logger/pom.xml b/osgi-bundles/bundles/logger/pom.xml index 8c08033a1..c2c1d9c09 100644 --- a/osgi-bundles/bundles/logger/pom.xml +++ b/osgi-bundles/bundles/logger/pom.xml @@ -44,6 +44,20 @@ com.google.code.findbugs jsr305 + + + com.google.inject + guice + + + + com.typesafe + config + + + jakarta.annotation + jakarta.annotation-api + jakarta.servlet jakarta.servlet-api diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java index 26989abdd..8b29c77cc 100644 --- a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java +++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/Activator.java @@ -44,6 +44,7 @@ public class Activator extends KillbillActivatorBase { @Override public void start(final BundleContext context) throws Exception { + super.start(context); loggerFactory = new KillbillLoggerFactory(context.getBundle()); logEntriesManager = new LogEntriesManager(); killbillLogListener = new KillbillLogWriter(logEntriesManager, loggerFactory); @@ -58,7 +59,7 @@ public void start(final BundleContext context) throws Exception { registrar = new OSGIKillbillRegistrar(); final PluginApp pluginApp = new PluginAppBuilder(PLUGIN_NAME).build(); - logsSseHandler = new LogsSseHandler(logEntriesManager); + logsSseHandler = new LogsSseHandler(logEntriesManager, killbillAPI.getRecordIdApi()); pluginApp.sse("/", logsSseHandler); final HttpServlet httpServlet = PluginApp.createServlet(pluginApp); registerServlet(context, httpServlet); diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogEntriesFilter.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogEntriesFilter.java new file mode 100644 index 000000000..36b5b359b --- /dev/null +++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogEntriesFilter.java @@ -0,0 +1,202 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +import org.jooby.Request; +import org.killbill.billing.ObjectType; +import org.killbill.billing.plugin.api.PluginTenantContext; +import org.killbill.billing.util.api.RecordIdApi; +import org.killbill.billing.util.callcontext.TenantContext; +import org.killbill.commons.utils.annotation.VisibleForTesting; +import org.slf4j.Logger; +import org.slf4j.LoggerFactory; + +import jakarta.annotation.Nonnull; + +// http://localhost:8080/killbill-osgi-logger/ -> no filter applied +// +// http://localhost:8080/killbill-osgi-logger/?accountId -> no filter applied +// http://localhost:8080/killbill-osgi-logger/?accountId= -> no filter applied +// http://localhost:8080/killbill-osgi-logger/?accountId=somename -> filter by accountId. Return if matched. If nothing found, then do not return anything (heartbeat) +// +// http://localhost:8080/killbill-osgi-logger/?userToken -> no filter +// http://localhost:8080/killbill-osgi-logger/?userToken= -> no filter +// http://localhost:8080/killbill-osgi-logger/?userToken=someToken -> filter by userToken. Return if matched. If nothing found, then do not return anything (heartbeat) +// +// http://localhost:8080/killbill-osgi-logger/?accountId&userToken -> no filter applied +// http://localhost:8080/killbill-osgi-logger/?accountId=&userToken -> no filter applied +// http://localhost:8080/killbill-osgi-logger/?accountId&userToken= -> no filter applied +// http://localhost:8080/killbill-osgi-logger/?accountId=&userToken= -> no filter applied +// +// http://localhost:8080/killbill-osgi-logger/?accountId=somename&userToken -> filter by accountId only. Return if matched. If nothing found, then do not return anything (heartbeat) +// http://localhost:8080/killbill-osgi-logger/?accountId=somename&userToken= -> filter by accountId only. Return if matched. If nothing found, then do not return anything (heartbeat) +// +// http://localhost:8080/killbill-osgi-logger/?accountId&userToken=someToken -> filter by userToken only. Return if matched. If nothing found, then do not return anything (heartbeat) +// http://localhost:8080/killbill-osgi-logger/?accountId=&userToken=someToken -> filter by userToken only. Return if matched. If nothing found, then do not return anything (heartbeat) +// +// http://localhost:8080/killbill-osgi-logger/?accountId=somename&userToken=someToken -> filter by accountId OR userToken: +// - if by accountId found, return it. +// - if by accountId not found but by userToken found, return it. +// - If nothing found, then do not return anything (heartbeat) +class LogEntriesFilter { + + private static final Logger logger = LoggerFactory.getLogger(LogEntriesFilter.class); + + private final AccountRecordIdValue accountRecordId; + private final Optional userToken; + private boolean noResult; + + LogEntriesFilter(final Request request, final RecordIdApi recordIdApi) { + // resolveAccountRecordId() very likely call to db. Doing this at apply() time will cause it called every + // SSE calls. + this.accountRecordId = resolveAccountRecordId(request, recordIdApi); + this.userToken = getNonBlankParam(request, "userToken"); + this.noResult = true; + } + + public boolean hasNoResult() { + return noResult; + } + + /** + * Filters entries against the resolved criteria. + * Returns matching entries; if no filter is active, returns all. + */ + public Iterable apply(final Iterable entries) { + // Reset per drain cycle so heartbeat logic in the handler works correctly + noResult = true; + + if (!accountRecordId.isRequested() && userToken.isEmpty()) { + noResult = false; + return entries; + } + + final List result = new ArrayList<>(); + for (final LogEntryJson entry : entries) { + if (match(entry)) { + result.add(entry); + noResult = false; + } + } + return result; + } + + /** + * OR logic: entry passes if it matches ANY active filter. + * If no filter is active, all entries pass. + */ + @VisibleForTesting + boolean match(final LogEntryJson entry) { + if (!accountRecordId.isRequested() && userToken.isEmpty()) { + return true; + } + if (accountRecordId.matches(entry.getAccountRecordId())) { + return true; + } + return userToken.isPresent() && userToken.get().equals(entry.getUserToken()); + } + + /** + * Resolves the {@code accountId} query param to an internal record ID. + * Returns empty if param is missing or blank. + * Returns a sentinel value if param is provided but unresolvable (invalid UUID or account not found), + * so the filter stays active and matches nothing. + */ + @VisibleForTesting + static AccountRecordIdValue resolveAccountRecordId(final Request request, final RecordIdApi recordIdApi) { + final Optional param = getNonBlankParam(request, "accountId"); + logger.debug("param.isPresent(): {}, value: {} ", param.isPresent(), param.orElse("")); + if (param.isEmpty()) { + return AccountRecordIdValue.empty(); + } + + try { + final UUID accountId = UUID.fromString(param.get()); + logger.debug("accountId UUID constructed"); + final TenantContext context = new PluginTenantContext(accountId, null); + final Long recordId = recordIdApi.getRecordId(accountId, ObjectType.ACCOUNT, context); + logger.debug("recordId: {}", recordId); + if (recordId != null && recordId > 0L) { + return AccountRecordIdValue.exist(recordId.toString()); + } + } catch (final IllegalArgumentException ignored) { + // Invalid UUID format + logger.info("receive invalid UUID as request parameter"); + } + + // accountId was provided but accountRecordId unresolvable. keep filter active with a value that never matches + return AccountRecordIdValue.notMatch(); + } + + /** + * Returns the param value only if it is present and non-blank. + * Treats missing, empty, or whitespace-only values as "not provided". + */ + @VisibleForTesting + static Optional getNonBlankParam(final Request request, final String name) { + final Optional value = request.param(name).toOptional(); + if (value.isPresent() && !value.get().isBlank()) { + return value; + } + return Optional.empty(); + } + + static class AccountRecordIdValue { + + enum State { + EMPTY, // SSE request parameter not set + NOT_MATCH, // accountId not found + EXIST + } + + private final State state; + private final String value; + + AccountRecordIdValue(final State state, final String value) { + this.state = state; + this.value = value; + } + + static AccountRecordIdValue empty() { + return new AccountRecordIdValue(State.EMPTY, null); + } + + static AccountRecordIdValue notMatch() { + return new AccountRecordIdValue(State.NOT_MATCH, null); + } + + static AccountRecordIdValue exist(@Nonnull final String value) { + return new AccountRecordIdValue(State.EXIST, value); + } + + /** True when the client provided a non-blank accountId param. */ + boolean isRequested() { + return state != State.EMPTY; + } + + /** True only when the accountId was resolved and its record ID equals the given value. */ + boolean matches(final String entryAccountRecordId) { + return state == State.EXIST && value.equals(entryAccountRecordId); + } + } +} diff --git a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java index c2e90d36a..a3d381dc5 100644 --- a/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java +++ b/osgi-bundles/bundles/logger/src/main/java/org/killbill/billing/osgi/bundles/logger/LogsSseHandler.java @@ -30,7 +30,9 @@ import org.jooby.Request; import org.jooby.Sse; import org.jooby.funzy.Throwing; +import org.killbill.billing.util.api.RecordIdApi; import org.killbill.commons.concurrent.Executors; +import org.killbill.commons.utils.collect.Iterables; public class LogsSseHandler implements Sse.Handler, Closeable { @@ -38,10 +40,12 @@ public class LogsSseHandler implements Sse.Handler, Closeable { private static final long PERIOD_MS = 1000; private final LogEntriesManager logEntriesManager; + private final RecordIdApi recordIdApi; private final ScheduledExecutorService scheduledExecutorService; - public LogsSseHandler(final LogEntriesManager logEntriesManager) { + public LogsSseHandler(final LogEntriesManager logEntriesManager, final RecordIdApi recordIdApi) { this.logEntriesManager = logEntriesManager; + this.recordIdApi = recordIdApi; this.scheduledExecutorService = Executors.newSingleThreadScheduledExecutor("LogsSseHandler"); } @@ -57,22 +61,35 @@ public void handle(final Request req, final Sse sse) { final UUID cacheId = UUID.fromString(sse.id()); logEntriesManager.subscribe(cacheId, lastEventId); + final LogEntriesFilter entriesFilter = new LogEntriesFilter(req, this.recordIdApi); + final AtomicReference lastLogId = new AtomicReference<>(); final ScheduledFuture future = scheduledExecutorService.scheduleAtFixedRate(new Runnable() { @Override public void run() { final Iterable logEntries = logEntriesManager.drain(cacheId); - if (!logEntries.iterator().hasNext()) { - // In case we have nothing to send, send a heartbeat to verify the client is still around - // That way, we can more quickly cleanup our subscriptions - // Note that we set the id as the last log id, so that we can easily resume - sse.event("heartbeat").id(lastLogId.get()).send(); + if (Iterables.isEmpty(logEntries)) { + sendHeartbeat(); } else { - for (final LogEntryJson logEntryJson : logEntries) { + for (final LogEntryJson logEntryJson : entriesFilter.apply(logEntries)) { sse.event(logEntryJson).id(logEntryJson.getId()).send(); lastLogId.set(logEntryJson.getId()); } + // Filtered clients may never match any entry. Without a periodic write, + // proxies/load balancers may kill idle connections and server won't detect dead peers. + if (entriesFilter.hasNoResult()) { + sendHeartbeat(); + } + } + } + + void sendHeartbeat() { + // send heartbeat. ID, if any, is the last log id so resume works correctly. + final Sse.Event heartbeat = sse.event("heartbeat"); + if (lastLogId.get() != null) { + heartbeat.id(lastLogId.get()); } + heartbeat.send(); } }, INITIAL_DELAY_MS, PERIOD_MS, TimeUnit.MILLISECONDS); diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogEntriesFilter.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogEntriesFilter.java new file mode 100644 index 000000000..f6d92d2e6 --- /dev/null +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogEntriesFilter.java @@ -0,0 +1,329 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.util.ArrayList; +import java.util.List; +import java.util.Optional; +import java.util.UUID; + +import org.jooby.Mutant; +import org.jooby.Request; +import org.killbill.billing.ObjectType; +import org.killbill.billing.util.api.RecordIdApi; +import org.killbill.commons.utils.collect.Iterables; +import org.mockito.Mockito; +import org.osgi.service.log.LogService; +import org.testng.Assert; +import org.testng.annotations.Test; + +public class TestLogEntriesFilter { + + // -- match() tests -- + + @Test(groups = "fast") + public void testMatchNoFilters() { + final LogEntriesFilter filter = buildFilter(null, null); + Assert.assertTrue(filter.match(makeEntry("msg", "token-1", "42"))); + } + + @Test(groups = "fast") + public void testMatchByUserTokenOnly() { + final LogEntriesFilter filter = buildFilter(null, "token-1"); + Assert.assertTrue(filter.match(makeEntry("msg", "token-1", "42"))); + Assert.assertFalse(filter.match(makeEntry("msg", "other", "42"))); + } + + @Test(groups = "fast") + public void testMatchByAccountRecordIdOnly() { + final UUID accountId = UUID.randomUUID(); + final LogEntriesFilter filter = buildFilterWithAccount(accountId, 42L, null); + Assert.assertTrue(filter.match(makeEntry("msg", "token-1", "42"))); + Assert.assertFalse(filter.match(makeEntry("msg", "token-1", "99"))); + } + + @Test(groups = "fast") + public void testMatchBothFiltersOr() { + final UUID accountId = UUID.randomUUID(); + final LogEntriesFilter filter = buildFilterWithAccount(accountId, 42L, "token-1"); + // Both match + Assert.assertTrue(filter.match(makeEntry("msg", "token-1", "42"))); + // accountRecordId matches only + Assert.assertTrue(filter.match(makeEntry("msg", "wrong", "42"))); + // userToken matches only + Assert.assertTrue(filter.match(makeEntry("msg", "token-1", "99"))); + // Neither matches + Assert.assertFalse(filter.match(makeEntry("msg", "wrong", "99"))); + } + + @Test(groups = "fast") + public void testMatchEntryWithNullFields() { + final LogEntriesFilter noFilter = buildFilter(null, null); + Assert.assertTrue(noFilter.match(makeEntry("msg", null, null))); + + final LogEntriesFilter tokenFilter = buildFilter(null, "any"); + Assert.assertFalse(tokenFilter.match(makeEntry("msg", null, null))); + + final UUID accountId = UUID.randomUUID(); + final LogEntriesFilter accountFilter = buildFilterWithAccount(accountId, 42L, null); + Assert.assertFalse(accountFilter.match(makeEntry("msg", null, null))); + } + + // -- getNonBlankParam() tests -- + + @Test(groups = "fast") + public void testGetNonBlankParamPresent() { + final Request request = mockRequestWithParam("userToken", "abc"); + Assert.assertEquals(LogEntriesFilter.getNonBlankParam(request, "userToken"), Optional.of("abc")); + } + + @Test(groups = "fast") + public void testGetNonBlankParamEmpty() { + final Request request = mockRequestWithParam("userToken", ""); + Assert.assertEquals(LogEntriesFilter.getNonBlankParam(request, "userToken"), Optional.empty()); + } + + @Test(groups = "fast") + public void testGetNonBlankParamBlank() { + final Request request = mockRequestWithParam("userToken", " "); + Assert.assertEquals(LogEntriesFilter.getNonBlankParam(request, "userToken"), Optional.empty()); + } + + @Test(groups = "fast") + public void testGetNonBlankParamMissing() { + final Request request = mockRequestWithParam("userToken", null); + Assert.assertEquals(LogEntriesFilter.getNonBlankParam(request, "userToken"), Optional.empty()); + } + + // -- resolveAccountRecordId() tests -- + + @Test(groups = "fast") + public void testResolveAccountRecordIdValid() { + final UUID accountId = UUID.randomUUID(); + final Request request = mockRequestWithParam("accountId", accountId.toString()); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(42L); + + final LogEntriesFilter.AccountRecordIdValue result = LogEntriesFilter.resolveAccountRecordId(request, recordIdApi); + Assert.assertTrue(result.isRequested()); + Assert.assertTrue(result.matches("42")); + Assert.assertFalse(result.matches("99")); + } + + @Test(groups = "fast") + public void testResolveAccountRecordIdNotFound() { + final UUID accountId = UUID.randomUUID(); + final Request request = mockRequestWithParam("accountId", accountId.toString()); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(null); + + final LogEntriesFilter.AccountRecordIdValue result = LogEntriesFilter.resolveAccountRecordId(request, recordIdApi); + Assert.assertTrue(result.isRequested(), "Filter should be active for provided but unresolvable accountId"); + Assert.assertFalse(result.matches("42"), "Unresolvable accountId should never match"); + } + + @Test(groups = "fast") + public void testResolveAccountRecordIdZero() { + final UUID accountId = UUID.randomUUID(); + final Request request = mockRequestWithParam("accountId", accountId.toString()); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(0L); + + final LogEntriesFilter.AccountRecordIdValue result = LogEntriesFilter.resolveAccountRecordId(request, recordIdApi); + Assert.assertTrue(result.isRequested()); + Assert.assertFalse(result.matches("0")); + } + + @Test(groups = "fast") + public void testResolveAccountRecordIdInvalidUuid() { + final Request request = mockRequestWithParam("accountId", "not-a-uuid"); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + + final LogEntriesFilter.AccountRecordIdValue result = LogEntriesFilter.resolveAccountRecordId(request, recordIdApi); + Assert.assertTrue(result.isRequested(), "Filter should be active for invalid UUID"); + Assert.assertFalse(result.matches("42")); + Mockito.verifyNoInteractions(recordIdApi); + } + + @Test(groups = "fast") + public void testResolveAccountRecordIdBlankParam() { + final Request request = mockRequestWithParam("accountId", ""); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + + final LogEntriesFilter.AccountRecordIdValue result = LogEntriesFilter.resolveAccountRecordId(request, recordIdApi); + Assert.assertFalse(result.isRequested(), "Blank param means not requested"); + Mockito.verifyNoInteractions(recordIdApi); + } + + // -- apply() tests -- + + @Test(groups = "fast") + public void testApplyNoFilter() { + final Request request = mockRequestWithParams(null, null); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + final List entries = List.of( + makeEntry("msg1", "token-a", "10"), + makeEntry("msg2", null, null)); + + final Iterable result = filter.apply(entries); + Assert.assertSame(result, entries, "No filter should return same iterable"); + Assert.assertFalse(filter.hasNoResult()); + } + + @Test(groups = "fast") + public void testApplyFilterByUserToken() { + final Request request = mockRequestWithParams(null, "my-token"); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + final List entries = List.of( + makeEntry("match", "my-token", null), + makeEntry("skip", "other", null), + makeEntry("skip2", null, null)); + + final List result = Iterables.toUnmodifiableList(filter.apply(entries)); + Assert.assertEquals(result.size(), 1); + Assert.assertEquals(result.get(0).getMessage(), "match"); + Assert.assertFalse(filter.hasNoResult()); + } + + @Test(groups = "fast") + public void testApplyFilterNothingMatches() { + final Request request = mockRequestWithParams(null, "non-existent"); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + final List entries = List.of( + makeEntry("msg1", "other", null), + makeEntry("msg2", null, null)); + + final List result = Iterables.toUnmodifiableList(filter.apply(entries)); + Assert.assertTrue(result.isEmpty()); + Assert.assertTrue(filter.hasNoResult()); + } + + @Test(groups = "fast") + public void testApplyHasNoResultResetsPerCycle() { + final Request request = mockRequestWithParams(null, "my-token"); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + // Cycle 1: match + filter.apply(List.of(makeEntry("match", "my-token", null))); + Assert.assertFalse(filter.hasNoResult()); + + // Cycle 2: no match — hasNoResult must reset to true + filter.apply(List.of(makeEntry("skip", "other", null))); + Assert.assertTrue(filter.hasNoResult()); + + // Cycle 3: match again + filter.apply(List.of(makeEntry("match2", "my-token", null))); + Assert.assertFalse(filter.hasNoResult()); + } + + @Test(groups = "fast") + public void testApplyOrLogicBothFilters() { + final UUID accountId = UUID.randomUUID(); + final Request request = mockRequestWithParams(accountId.toString(), "my-token"); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(42L); + + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + final List entries = List.of( + makeEntry("both match", "my-token", "42"), + makeEntry("token only", "my-token", "99"), + makeEntry("account only", "other", "42"), + makeEntry("neither", "other", "99")); + + final List result = Iterables.toUnmodifiableList(filter.apply(entries)); + Assert.assertEquals(result.size(), 3); + Assert.assertEquals(result.get(0).getMessage(), "both match"); + Assert.assertEquals(result.get(1).getMessage(), "token only"); + Assert.assertEquals(result.get(2).getMessage(), "account only"); + Assert.assertFalse(filter.hasNoResult()); + } + + @Test(groups = "fast") + public void testApplyUnresolvableAccountIdMatchesNothing() { + final UUID unknownAccountId = UUID.randomUUID(); + final Request request = mockRequestWithParams(unknownAccountId.toString(), null); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(unknownAccountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(null); + + final LogEntriesFilter filter = new LogEntriesFilter(request, recordIdApi); + + final List entries = List.of( + makeEntry("msg1", "token", "42"), + makeEntry("msg2", null, null)); + + // Unresolvable accountId → sentinel keeps filter active → nothing matches + final List result = Iterables.toUnmodifiableList(filter.apply(entries)); + Assert.assertTrue(result.isEmpty()); + Assert.assertTrue(filter.hasNoResult()); + } + + // -- Helpers -- + + private static LogEntryJson makeEntry(final String message, final String userToken, final String accountRecordId) { + return new LogEntryJson(null, LogService.LOG_INFO, "test.Logger", message, userToken, null, accountRecordId, null); + } + + private static LogEntriesFilter buildFilter(final String accountId, final String userToken) { + final Request request = mockRequestWithParams(accountId, userToken); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + return new LogEntriesFilter(request, recordIdApi); + } + + private static LogEntriesFilter buildFilterWithAccount(final UUID accountId, final Long recordId, final String userToken) { + final Request request = mockRequestWithParams(accountId.toString(), userToken); + final RecordIdApi recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(accountId), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(recordId); + return new LogEntriesFilter(request, recordIdApi); + } + + private static Request mockRequestWithParam(final String name, final String value) { + final Request request = Mockito.mock(Request.class); + final Mutant mutant = Mockito.mock(Mutant.class); + Mockito.when(mutant.toOptional()).thenReturn(value == null ? Optional.empty() : Optional.of(value)); + Mockito.when(request.param(name)).thenReturn(mutant); + return request; + } + + private static Request mockRequestWithParams(final String accountId, final String userToken) { + final Request request = Mockito.mock(Request.class); + + final Mutant accountMutant = Mockito.mock(Mutant.class); + Mockito.when(accountMutant.toOptional()).thenReturn(accountId == null ? Optional.empty() : Optional.of(accountId)); + Mockito.when(request.param("accountId")).thenReturn(accountMutant); + + final Mutant tokenMutant = Mockito.mock(Mutant.class); + Mockito.when(tokenMutant.toOptional()).thenReturn(userToken == null ? Optional.empty() : Optional.of(userToken)); + Mockito.when(request.param("userToken")).thenReturn(tokenMutant); + + return request; + } +} diff --git a/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java new file mode 100644 index 000000000..dec3b2c22 --- /dev/null +++ b/osgi-bundles/bundles/logger/src/test/java/org/killbill/billing/osgi/bundles/logger/TestLogsSseHandler.java @@ -0,0 +1,581 @@ +/* + * Copyright 2020-2026 Equinix, Inc + * Copyright 2014-2026 The Billing Project, LLC + * + * The Billing Project licenses this file to you under the Apache License, version 2.0 + * (the "License"); you may not use this file except in compliance with the + * License. You may obtain a copy of the License at: + * + * http://www.apache.org/licenses/LICENSE-2.0 + * + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, WITHOUT + * WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the + * License for the specific language governing permissions and limitations + * under the License. + */ + +package org.killbill.billing.osgi.bundles.logger; + +import java.io.BufferedReader; +import java.io.InputStream; +import java.io.InputStreamReader; +import java.net.ServerSocket; +import java.net.URI; +import java.net.http.HttpClient; +import java.net.http.HttpRequest; +import java.net.http.HttpResponse; +import java.time.Duration; +import java.util.Map; +import java.util.UUID; +import java.util.concurrent.ExecutorService; +import java.util.concurrent.Executors; +import java.util.concurrent.Future; +import java.util.concurrent.TimeUnit; +import java.util.stream.Collectors; + +import org.jooby.Env; +import org.jooby.Jooby; +import org.jooby.jetty.Jetty; +import org.jooby.json.Jackson; +import org.killbill.billing.ObjectType; +import org.killbill.billing.util.api.RecordIdApi; +import org.osgi.service.log.LogService; +import org.mockito.Mockito; +import org.testng.Assert; +import org.testng.annotations.AfterMethod; +import org.testng.annotations.BeforeMethod; +import org.testng.annotations.Test; + +import com.google.inject.Binder; +import com.typesafe.config.Config; +import com.typesafe.config.ConfigFactory; +import com.typesafe.config.ConfigValueFactory; + +public class TestLogsSseHandler { + + private static final UUID ACCOUNT_ID = UUID.randomUUID(); + private static final Long ACCOUNT_RECORD_ID = 42L; + + private Jooby app; + private LogEntriesManager logEntriesManager; + private LogsSseHandler logsSseHandler; + private RecordIdApi recordIdApi; + private int port; + + @BeforeMethod(groups = "slow") + public void setUp() throws Exception { + port = findFreePort(); + logEntriesManager = new LogEntriesManager(); + recordIdApi = Mockito.mock(RecordIdApi.class); + Mockito.when(recordIdApi.getRecordId(Mockito.eq(ACCOUNT_ID), Mockito.eq(ObjectType.ACCOUNT), Mockito.any())) + .thenReturn(ACCOUNT_RECORD_ID); + logsSseHandler = new LogsSseHandler(logEntriesManager, recordIdApi); + + final Config testConfig = ConfigFactory.empty("test-config") + .withValue("server.join", ConfigValueFactory.fromAnyRef(false)) + .withValue("application.port", ConfigValueFactory.fromAnyRef(port)) + .withValue("server.module", ConfigValueFactory.fromAnyRef(Jetty.class.getName())); + + app = new Jooby() {{ + use(new Jackson()); + sse("/", logsSseHandler); + }}; + + app.use(new Jooby.Module() { + @Override + public void configure(final Env env, final Config config, final Binder binder) { + } + + @Override + public Config config() { + return testConfig; + } + }); + + app.start(); + } + + @AfterMethod(groups = "slow") + public void tearDown() throws Exception { + if (logsSseHandler != null) { + logsSseHandler.close(); + } + if (app != null) { + app.stop(); + } + } + + @Test(groups = "slow") + public void testReceivesLogEntriesBeforeConnect() throws Exception { + recordErrorEvent("com.acme.MyService", "Something broke", "user-token-123", "1", "2"); + recordInfoEvent("com.acme.OtherService", "All good here", null, null, null); + + final HttpResponse response = connectSse(); + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("Something broke") && sb.toString().contains("All good here")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("Something broke"), "Expected 'Something broke' in SSE output"); + Assert.assertTrue(sseOutput.contains("All good here"), "Expected 'All good here' in SSE output"); + Assert.assertTrue(sseOutput.contains("ERROR"), "Expected 'ERROR' level in SSE output"); + Assert.assertTrue(sseOutput.contains("INFO"), "Expected 'INFO' level in SSE output"); + Assert.assertTrue(sseOutput.contains("user-token-123"), "Expected userToken in SSE output"); + Assert.assertFalse(sseOutput.contains("I never recorded in event")); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testReceivesLiveLogEntries() throws Exception { + final HttpResponse response = connectSse(); + Thread.sleep(1000); // wait for SSE subscription to be established + + recordWarningEvent("com.acme.LiveService", "Live event arrived", null, null, null); + + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("Live event arrived")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("Live event arrived"), "Expected live event in SSE output"); + Assert.assertTrue(sseOutput.contains("WARNING"), "Expected 'WARNING' level in SSE output"); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testFilterByAccountId() throws Exception { + recordErrorEvent("com.acme.ServiceA", "Matching account event", null, null, ACCOUNT_RECORD_ID.toString()); + recordErrorEvent("com.acme.ServiceB", "Other account event", null, null, "999"); + recordInfoEvent("com.acme.ServiceC", "No account event", null, null, null); + + final HttpResponse response = connectSse(Map.of("accountId", ACCOUNT_ID.toString())); + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("Matching account event")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("Matching account event")); + Assert.assertFalse(sseOutput.contains("Other account event")); + Assert.assertFalse(sseOutput.contains("No account event")); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testFilterByUserToken() throws Exception { + recordErrorEvent("com.acme.ServiceA", "Token match event", "my-token", null, null); + recordErrorEvent("com.acme.ServiceB", "Other token event", "other-token", null, null); + recordInfoEvent("com.acme.ServiceC", "No token event", null, null, null); + + final HttpResponse response = connectSse(Map.of("userToken", "my-token")); + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("Token match event")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("Token match event")); + Assert.assertFalse(sseOutput.contains("Other token event")); + Assert.assertFalse(sseOutput.contains("No token event")); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testFilterByAccountIdOrUserToken() throws Exception { + // OR logic: entry passes if it matches accountId OR userToken + recordErrorEvent("com.acme.ServiceA", "Account match", null, null, ACCOUNT_RECORD_ID.toString()); + recordErrorEvent("com.acme.ServiceB", "Token match", "my-token", null, "999"); + recordInfoEvent("com.acme.ServiceC", "Neither match", "other-token", null, "888"); + + final HttpResponse response = connectSse(Map.of( + "accountId", ACCOUNT_ID.toString(), + "userToken", "my-token")); + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("Account match") && sb.toString().contains("Token match")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("Account match"), "Expected account-matched entry (OR)"); + Assert.assertTrue(sseOutput.contains("Token match"), "Expected token-matched entry (OR)"); + Assert.assertFalse(sseOutput.contains("Neither match"), "Should not contain unrelated entry"); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testHeartbeatSentWhenAllEntriesFilteredOut() throws Exception { + // Record entries that will NOT match the filter + recordErrorEvent("com.acme.ServiceA", "Unrelated event", "other-token", null, "999"); + recordInfoEvent("com.acme.ServiceB", "Another unrelated", null, null, null); + + final HttpResponse response = connectSse(Map.of("userToken", "non-existent-token")); + final ExecutorService reader = Executors.newSingleThreadExecutor(); + try { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (sb.toString().contains("heartbeat")) { + break; + } + } + } + return sb.toString(); + }); + + final String sseOutput = future.get(10, TimeUnit.SECONDS); + + Assert.assertTrue(sseOutput.contains("heartbeat"), "Expected heartbeat when no entries match filter"); + Assert.assertFalse(sseOutput.contains("Unrelated event"), "Filtered entries should not appear"); + Assert.assertFalse(sseOutput.contains("Another unrelated"), "Filtered entries should not appear"); + } finally { + reader.shutdownNow(); + response.body().close(); + } + } + + @Test(groups = "slow") + public void testFullSseLifecycle() throws Exception { + // -- Step 1-3: Connect without filter or Last-Event-ID. Read initial entries. -- + recordInfoEvent("com.acme.Init", "Entry one", null, null, null); + recordInfoEvent("com.acme.Init", "Entry two", null, null, null); + + String lastReceivedId; + try (final SseClientForTesting session = SseClientForTesting.connect(port)) { + final String output = session.readUntil("Entry one", "Entry two"); + Assert.assertTrue(output.contains("Entry one")); + Assert.assertTrue(output.contains("Entry two")); + + // Extract the last event id from the SSE stream for reconnection later + lastReceivedId = extractLastEventId(output); + Assert.assertNotNull(lastReceivedId, "Should have received at least one event id"); + } + + // -- Step 4: Server adds entry with userToken -- + recordErrorEvent("com.acme.Auth", "Token event", "request-abc", null, null); + + // -- Step 5: Filter by non-matching userToken -> heartbeat only -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of("userToken", "non-existent"))) { + final String output = session.readUntil("heartbeat"); + Assert.assertTrue(output.contains("heartbeat")); + Assert.assertFalse(output.contains("Token event")); + Assert.assertFalse(output.contains("Entry one")); + } + + // -- Step 6: Filter by matching userToken -> match -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of("userToken", "request-abc"))) { + final String output = session.readUntil("Token event"); + Assert.assertTrue(output.contains("Token event")); + Assert.assertFalse(output.contains("Entry one"), "Non-matching entries should be filtered"); + } + + // -- Step 7: Server adds entry with accountRecordId -- + recordWarningEvent("com.acme.Billing", "Account event", null, null, ACCOUNT_RECORD_ID.toString()); + + // -- Step 8a: Filter by non-matching accountId -> heartbeat only -- + final UUID wrongAccountId = UUID.randomUUID(); + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of("accountId", wrongAccountId.toString()))) { + final String output = session.readUntil("heartbeat"); + Assert.assertTrue(output.contains("heartbeat")); + Assert.assertFalse(output.contains("Account event")); + } + + // -- Step 8b: Filter by matching accountId -> match -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of("accountId", ACCOUNT_ID.toString()))) { + final String output = session.readUntil("Account event"); + Assert.assertTrue(output.contains("Account event")); + Assert.assertFalse(output.contains("Entry one"), "Non-matching entries should be filtered"); + } + + // -- Step 9: Filter with both userToken AND accountId (OR logic) -> either match passes -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of( + "accountId", ACCOUNT_ID.toString(), + "userToken", "request-abc"))) { + final String output = session.readUntil("Token event", "Account event"); + Assert.assertTrue(output.contains("Token event"), "userToken-matched entry should pass (OR)"); + Assert.assertTrue(output.contains("Account event"), "accountRecordId-matched entry should pass (OR)"); + Assert.assertFalse(output.contains("Entry one"), "Non-matching entries should be filtered"); + } + + // -- Step 10: Filter with no match -> heartbeat only -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of("userToken", "totally-unknown"))) { + final String output = session.readUntil("heartbeat"); + Assert.assertTrue(output.contains("heartbeat")); + Assert.assertFalse(output.contains("Token event")); + Assert.assertFalse(output.contains("Account event")); + Assert.assertFalse(output.contains("Entry one")); + } + + // -- Step 11: Server adds more entries -- + recordInfoEvent("com.acme.New", "Fresh entry after resume point", null, null, null); + + // -- Step 12: Reconnect with Last-Event-ID (no filter) -> only entries after that ID -- + try (final SseClientForTesting session = SseClientForTesting.connect(port, Map.of(), lastReceivedId)) { + final String output = session.readUntil("Fresh entry after resume point"); + Assert.assertTrue(output.contains("Fresh entry after resume point"), "New entry should be received"); + Assert.assertTrue(output.contains("Token event"), "Entry added after resume point should be received"); + Assert.assertTrue(output.contains("Account event"), "Entry added after resume point should be received"); + Assert.assertFalse(output.contains("Entry one"), "Entry before Last-Event-ID should not be received"); + Assert.assertFalse(output.contains("Entry two"), "Entry before Last-Event-ID should not be received"); + } + } + + // -- Helpers -- + + private void recordInfoEvent(final String loggerName, + final String message, + final String userToken, + final String tenantRecordId, + final String accountRecordId) { + recordEvent(LogService.LOG_INFO, loggerName, message, userToken, tenantRecordId, accountRecordId); + } + + private void recordWarningEvent(final String loggerName, + final String message, + final String userToken, + final String tenantRecordId, + final String accountRecordId) { + recordEvent(LogService.LOG_WARNING, loggerName, message, userToken, tenantRecordId, accountRecordId); + } + + private void recordErrorEvent(final String loggerName, + final String message, + final String userToken, + final String tenantRecordId, + final String accountRecordId) { + recordEvent(LogService.LOG_ERROR, loggerName, message, userToken, tenantRecordId, accountRecordId); + } + + private void recordEvent(final int level, + final String loggerName, + final String message, + final String userToken, + final String tenantRecordId, + final String accountRecordId) { + logEntriesManager.recordEvent(new LogEntryJson(null, level, loggerName, message, userToken, tenantRecordId, accountRecordId, null)); + } + + private HttpResponse connectSse(final Map queryParams) throws Exception { + final StringBuilder uriBuilder = new StringBuilder("http://localhost:").append(port).append("/"); + if (!queryParams.isEmpty()) { + final String query = queryParams.entrySet().stream() + .map(e -> e.getKey() + "=" + e.getValue()) + .collect(Collectors.joining("&")); + uriBuilder.append("?").append(query); + } + + final HttpClient client = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(5)) + .build(); + + final HttpRequest request = HttpRequest.newBuilder(URI.create(uriBuilder.toString())) + .header("Accept", "text/event-stream") + .GET() + .build(); + + final HttpResponse response = client.send(request, HttpResponse.BodyHandlers.ofInputStream()); + Assert.assertEquals(response.statusCode(), 200); + return response; + } + + private HttpResponse connectSse() throws Exception { + return connectSse(Map.of()); + } + + private static int findFreePort() throws Exception { + try (final ServerSocket socket = new ServerSocket(0)) { + return socket.getLocalPort(); + } + } + + /** + * Extracts the last "id:" value from raw SSE output. + * SSE format: "id:some-uuid\n" + */ + private static String extractLastEventId(final String sseOutput) { + String lastId = null; + for (final String line : sseOutput.split("\n")) { + if (line.startsWith("id:")) { + lastId = line.substring(3).trim(); + } + } + return lastId; + } +} + + +/** + * Lightweight SSE client for tests. Each instance represents one connection lifecycle: + * connect → read → close. + */ +class SseClientForTesting implements AutoCloseable { + + private final HttpResponse response; + private final ExecutorService reader; + + private SseClientForTesting(final HttpResponse response) { + this.response = response; + this.reader = Executors.newSingleThreadExecutor(); + } + + public static SseClientForTesting connect(final int port) throws Exception { + return connect(port, Map.of(), null); + } + + public static SseClientForTesting connect(final int port, final Map queryParams) throws Exception { + return connect(port, queryParams, null); + } + + public static SseClientForTesting connect(final int port, + final Map queryParams, + final String lastEventId) throws Exception { + final StringBuilder uriBuilder = new StringBuilder("http://localhost:").append(port).append("/"); + if (!queryParams.isEmpty()) { + final String query = queryParams.entrySet().stream() + .map(e -> e.getKey() + "=" + e.getValue()) + .collect(Collectors.joining("&")); + uriBuilder.append("?").append(query); + } + + final HttpRequest.Builder requestBuilder = HttpRequest.newBuilder(URI.create(uriBuilder.toString())) + .header("Accept", "text/event-stream") + .GET(); + + if (lastEventId != null) { + requestBuilder.header("Last-Event-ID", lastEventId); + } + + final HttpClient client = HttpClient.newBuilder() + .connectTimeout(Duration.ofSeconds(5)) + .build(); + + final HttpResponse response = client.send(requestBuilder.build(), HttpResponse.BodyHandlers.ofInputStream()); + if (response.statusCode() != 200) { + throw new IllegalStateException("SSE connection failed with status " + response.statusCode()); + } + + return new SseClientForTesting(response); + } + + /** + * Reads the SSE stream until all expected strings are found, or times out. + */ + public String readUntil(final String... expectedContents) throws Exception { + return readUntil(10, expectedContents); + } + + /** + * Reads the SSE stream until all expected strings are found, or times out. + */ + public String readUntil(final int timeoutSeconds, final String... expectedContents) throws Exception { + final Future future = reader.submit(() -> { + final StringBuilder sb = new StringBuilder(); + try (final BufferedReader br = new BufferedReader(new InputStreamReader(response.body()))) { + String line; + while ((line = br.readLine()) != null) { + sb.append(line).append('\n'); + if (containsAll(sb.toString(), expectedContents)) { + break; + } + } + } + return sb.toString(); + }); + return future.get(timeoutSeconds, TimeUnit.SECONDS); + } + + @Override + public void close() throws Exception { + reader.shutdownNow(); + response.body().close(); + } + + private static boolean containsAll(final String content, final String... expected) { + for (final String s : expected) { + if (!content.contains(s)) { + return false; + } + } + return true; + } +}