diff --git a/extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/DatabaseReaderFactory.java b/extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/DatabaseReaderFactory.java index 19155c86e8f..61b969eee6f 100644 --- a/extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/DatabaseReaderFactory.java +++ b/extra/modules/greenbids-real-time-data/src/main/java/org/prebid/server/hooks/modules/greenbids/real/time/data/config/DatabaseReaderFactory.java @@ -4,7 +4,6 @@ import com.maxmind.geoip2.DatabaseReader; import io.netty.handler.codec.http.HttpResponseStatus; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import io.vertx.core.file.FileSystem; import io.vertx.core.file.OpenOptions; @@ -45,11 +44,10 @@ public DatabaseReaderFactory(GreenbidsRealTimeDataProperties properties, Vertx v } @Override - public void initialize(Promise initializePromise) { - downloadAndExtract() + public Future initialize() { + return downloadAndExtract() .onSuccess(databaseReaderRef::set) - .mapEmpty() - .onComplete(initializePromise); + .mapEmpty(); } private Future downloadAndExtract() { diff --git a/src/main/java/org/prebid/server/analytics/reporter/agma/AgmaAnalyticsReporter.java b/src/main/java/org/prebid/server/analytics/reporter/agma/AgmaAnalyticsReporter.java index ed99c241ee5..48aa9e5d3d8 100644 --- a/src/main/java/org/prebid/server/analytics/reporter/agma/AgmaAnalyticsReporter.java +++ b/src/main/java/org/prebid/server/analytics/reporter/agma/AgmaAnalyticsReporter.java @@ -12,7 +12,6 @@ import io.vertx.core.AsyncResult; import io.vertx.core.Future; import io.vertx.core.MultiMap; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import io.vertx.core.http.HttpHeaders; import io.vertx.core.http.HttpMethod; @@ -97,9 +96,9 @@ public AgmaAnalyticsReporter(AgmaAnalyticsProperties agmaAnalyticsProperties, } @Override - public void initialize(Promise initializePromise) { + public Future initialize() { vertx.setPeriodic(bufferTimeoutMs, ignored -> sendEvents(buffer.pollAll())); - initializePromise.complete(); + return Future.succeededFuture(); } @Override diff --git a/src/main/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporter.java b/src/main/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporter.java index 1fd2ab7749b..b2c947c3a9c 100644 --- a/src/main/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporter.java +++ b/src/main/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporter.java @@ -1,7 +1,6 @@ package org.prebid.server.analytics.reporter.pubstack; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.apache.commons.collections4.MapUtils; import org.apache.commons.lang3.BooleanUtils; @@ -116,9 +115,9 @@ public String name() { } @Override - public void initialize(Promise initializePromise) { + public Future initialize() { vertx.setPeriodic(configurationRefreshDelay, id -> fetchRemoteConfig()); - fetchRemoteConfig().onComplete(initializePromise); + return fetchRemoteConfig(); } void shutdown() { diff --git a/src/main/java/org/prebid/server/currency/CurrencyConversionService.java b/src/main/java/org/prebid/server/currency/CurrencyConversionService.java index efcaf4ad7c2..ab398f877cd 100644 --- a/src/main/java/org/prebid/server/currency/CurrencyConversionService.java +++ b/src/main/java/org/prebid/server/currency/CurrencyConversionService.java @@ -2,7 +2,6 @@ import com.iab.openrtb.request.BidRequest; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.apache.commons.collections4.MapUtils; import org.apache.commons.lang3.BooleanUtils; @@ -68,7 +67,7 @@ public CurrencyConversionService(ExternalConversionProperties externalConversion * Must be called on Vertx event loop thread. */ @Override - public void initialize(Promise initializePromise) { + public Future initialize() { if (externalConversionProperties != null) { final Long refreshPeriod = externalConversionProperties.getRefreshPeriodMs(); final Long defaultTimeout = externalConversionProperties.getDefaultTimeoutMs(); @@ -84,7 +83,7 @@ public void initialize(Promise initializePromise) { externalConversionProperties.getMetrics().createCurrencyRatesGauge(this::isRatesStale); } - initializePromise.tryComplete(); + return Future.succeededFuture(); } /** diff --git a/src/main/java/org/prebid/server/settings/service/DatabasePeriodicRefreshService.java b/src/main/java/org/prebid/server/settings/service/DatabasePeriodicRefreshService.java index 7714243954d..9ace90196c2 100644 --- a/src/main/java/org/prebid/server/settings/service/DatabasePeriodicRefreshService.java +++ b/src/main/java/org/prebid/server/settings/service/DatabasePeriodicRefreshService.java @@ -1,7 +1,6 @@ package org.prebid.server.settings.service; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.apache.commons.lang3.StringUtils; import org.prebid.server.execution.timeout.Timeout; @@ -104,12 +103,12 @@ public DatabasePeriodicRefreshService(String initQuery, } @Override - public void initialize(Promise initializePromise) { + public Future initialize() { getAll(); if (refreshPeriod > 0) { vertx.setPeriodic(refreshPeriod, aLong -> refresh()); } - initializePromise.tryComplete(); + return Future.succeededFuture(); } private void getAll() { diff --git a/src/main/java/org/prebid/server/settings/service/HttpPeriodicRefreshService.java b/src/main/java/org/prebid/server/settings/service/HttpPeriodicRefreshService.java index 7669074d455..b32773eb7ce 100644 --- a/src/main/java/org/prebid/server/settings/service/HttpPeriodicRefreshService.java +++ b/src/main/java/org/prebid/server/settings/service/HttpPeriodicRefreshService.java @@ -4,7 +4,6 @@ import com.fasterxml.jackson.databind.JsonNode; import com.fasterxml.jackson.databind.node.ObjectNode; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.prebid.server.exception.PreBidException; import org.prebid.server.json.DecodeException; @@ -91,13 +90,13 @@ public HttpPeriodicRefreshService(String refreshUrl, } @Override - public void initialize(Promise initializePromise) { + public Future initialize() { getAll(); if (refreshPeriod > 0) { vertx.setPeriodic(refreshPeriod, aLong -> refresh()); } - initializePromise.tryComplete(); + return Future.succeededFuture(); } private void getAll() { diff --git a/src/main/java/org/prebid/server/settings/service/S3PeriodicRefreshService.java b/src/main/java/org/prebid/server/settings/service/S3PeriodicRefreshService.java index 35ff2a6f55a..30688dd09cf 100644 --- a/src/main/java/org/prebid/server/settings/service/S3PeriodicRefreshService.java +++ b/src/main/java/org/prebid/server/settings/service/S3PeriodicRefreshService.java @@ -2,7 +2,6 @@ import io.vertx.core.CompositeFuture; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.prebid.server.auction.model.Tuple2; import org.prebid.server.log.Logger; @@ -73,15 +72,13 @@ public S3PeriodicRefreshService(S3AsyncClient asyncClient, } @Override - public void initialize(Promise initializePromise) { - fetchStoredDataResult(clock.millis(), MetricName.initialize) - .mapEmpty() - .onComplete(initializePromise); - + public Future initialize() { if (refreshPeriod > 0) { logger.info("Starting s3 periodic refresh for " + cacheType + " every " + refreshPeriod + " s"); vertx.setPeriodic(refreshPeriod, ignored -> fetchStoredDataResult(clock.millis(), MetricName.update)); } + + return fetchStoredDataResult(clock.millis(), MetricName.initialize).mapEmpty(); } private Future> fetchStoredDataResult(long startTime, MetricName metricName) { diff --git a/src/main/java/org/prebid/server/util/system/CpuLoadAverageStats.java b/src/main/java/org/prebid/server/util/system/CpuLoadAverageStats.java index 6be897cd1d4..0ff182c4ac7 100644 --- a/src/main/java/org/prebid/server/util/system/CpuLoadAverageStats.java +++ b/src/main/java/org/prebid/server/util/system/CpuLoadAverageStats.java @@ -1,6 +1,6 @@ package org.prebid.server.util.system; -import io.vertx.core.Promise; +import io.vertx.core.Future; import io.vertx.core.Vertx; import org.prebid.server.vertx.Initializable; import oshi.SystemInfo; @@ -31,10 +31,10 @@ public CpuLoadAverageStats(Vertx vertx, long measurementIntervalMillis) { } @Override - public void initialize(Promise initializePromise) { + public Future initialize() { measureCpuLoad(); vertx.setPeriodic(measurementIntervalMillis, timerId -> measureCpuLoad()); - initializePromise.tryComplete(); + return Future.succeededFuture(); } private void measureCpuLoad() { diff --git a/src/main/java/org/prebid/server/vertx/Initializable.java b/src/main/java/org/prebid/server/vertx/Initializable.java index 5c12c30e47d..3ecfe6e8037 100644 --- a/src/main/java/org/prebid/server/vertx/Initializable.java +++ b/src/main/java/org/prebid/server/vertx/Initializable.java @@ -1,7 +1,7 @@ package org.prebid.server.vertx; +import io.vertx.core.Future; import io.vertx.core.Handler; -import io.vertx.core.Promise; /** * Denotes components requiring initialization after they have been created. @@ -12,5 +12,5 @@ @FunctionalInterface public interface Initializable { - void initialize(Promise initializePromise); + Future initialize(); } diff --git a/src/main/java/org/prebid/server/vertx/verticles/server/DaemonVerticle.java b/src/main/java/org/prebid/server/vertx/verticles/server/DaemonVerticle.java index c6175ccaafa..e64a3eab5d1 100644 --- a/src/main/java/org/prebid/server/vertx/verticles/server/DaemonVerticle.java +++ b/src/main/java/org/prebid/server/vertx/verticles/server/DaemonVerticle.java @@ -11,10 +11,8 @@ import org.prebid.server.vertx.CloseableAdapter; import org.prebid.server.vertx.Initializable; -import java.util.ArrayList; import java.util.Collection; import java.util.List; -import java.util.function.Consumer; import java.util.function.Function; public class DaemonVerticle extends AbstractVerticle { @@ -33,31 +31,26 @@ public DaemonVerticle(List initializables, List startPromise) { - all(initializables, initializable -> initializable::initialize).onComplete(startPromise); + all(initializables, Initializable::initialize, true).onComplete(startPromise); } @Override public void stop(Promise stopPromise) { - all(closeables, closeable -> closeable::close).onComplete(stopPromise); + all(closeables, closeable -> Future.future(closeable::close), false).onComplete(stopPromise); } - private static Future all(Collection entries, - Function>> entryToPromiseConsumerMapper) { + private static Future all( + Collection entries, + Function> entryToFutureMapper, + boolean start) { - final List> entriesFutures = new ArrayList<>(); - - for (E entry : entries) { - final Promise entryPromise = Promise.promise(); - entriesFutures.add(entryPromise.future()); - - entryToPromiseConsumerMapper.apply(entry).accept(entryPromise); - } - - return Future.all(entriesFutures) + return Future.all(entries.stream().map(entryToFutureMapper).toList()) .onSuccess(r -> logger.info( - "Successfully started {} instance on thread: {}", + "Successfully {} {} instance on thread: {}", + start ? "started" : "stopped", DaemonVerticle.class.getSimpleName(), Thread.currentThread().getName())) .mapEmpty(); } + } diff --git a/src/test/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporterTest.java b/src/test/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporterTest.java index e971498c598..55753587562 100644 --- a/src/test/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporterTest.java +++ b/src/test/java/org/prebid/server/analytics/reporter/pubstack/PubstackAnalyticsReporterTest.java @@ -2,7 +2,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -93,7 +92,7 @@ public void initializeShouldFetchConfigAndSetPeriodicTimerForConfigUpdate() thro Future.succeededFuture(HttpClientResponse.of(200, null, mapper.writeValueAsString(pubstackConfig)))); // when - pubstackAnalyticsReporter.initialize(Promise.promise()); + pubstackAnalyticsReporter.initialize(); // then verify(vertx).setPeriodic(anyLong(), any()); @@ -109,17 +108,16 @@ public void initializeShouldFailUpdateSendBuffersAndSetTimerWhenEndpointFromRemo throws JsonProcessingException { // given - final Promise promise = Promise.promise(); final PubstackConfig pubstackConfig = PubstackConfig.of("newScopeId", "invalid", Collections.singletonMap(EventType.auction, true)); given(httpClient.get(anyString(), anyLong())).willReturn( Future.succeededFuture(HttpClientResponse.of(200, null, mapper.writeValueAsString(pubstackConfig)))); // when - pubstackAnalyticsReporter.initialize(promise); + final Future result = pubstackAnalyticsReporter.initialize(); // then - assertThatThrownBy(() -> promise.future().await(5, TimeUnit.SECONDS)) + assertThatThrownBy(() -> result.await(5, TimeUnit.SECONDS)) .hasMessage("[pubstack] Failed to create event report url for endpoint: invalid") .isInstanceOf(PreBidException.class); verify(auctionHandler).reportEvents(); @@ -138,8 +136,8 @@ public void initializeShouldNotUpdateEventsIfFetchedConfigIsSameAsPrevious() thr Future.succeededFuture(HttpClientResponse.of(200, null, mapper.writeValueAsString(pubstackConfig)))); // when - pubstackAnalyticsReporter.initialize(Promise.promise()); - pubstackAnalyticsReporter.initialize(Promise.promise()); + pubstackAnalyticsReporter.initialize(); + pubstackAnalyticsReporter.initialize(); // then verify(httpClient, times(2)).get(anyString(), anyLong()); @@ -157,7 +155,7 @@ public void initializeShouldNotSendEventsAndUpdateConfigsWhenResponseStatusIsNot Future.succeededFuture(HttpClientResponse.of(400, null, null))); // when - pubstackAnalyticsReporter.initialize(Promise.promise()); + pubstackAnalyticsReporter.initialize(); // then verify(vertx).setPeriodic(anyLong(), any()); @@ -173,7 +171,7 @@ public void initializeShouldNotSendEventsAndUpdateConfigsWhenCantParseResponse() Future.succeededFuture(HttpClientResponse.of(200, null, "{\"endpoint\" : {}}"))); // when - pubstackAnalyticsReporter.initialize(Promise.promise()); + pubstackAnalyticsReporter.initialize(); // then verify(vertx).setPeriodic(anyLong(), any()); diff --git a/src/test/java/org/prebid/server/auction/CurrencyConversionServiceTest.java b/src/test/java/org/prebid/server/auction/CurrencyConversionServiceTest.java index 61a28789710..ed9030c493f 100644 --- a/src/test/java/org/prebid/server/auction/CurrencyConversionServiceTest.java +++ b/src/test/java/org/prebid/server/auction/CurrencyConversionServiceTest.java @@ -4,7 +4,6 @@ import com.iab.openrtb.request.BidRequest; import io.vertx.core.Future; import io.vertx.core.Handler; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -397,7 +396,7 @@ private CurrencyConversionService createInitializedService(String url, clock, jacksonMapper)); - currencyService.initialize(Promise.promise()); + currencyService.initialize(); return currencyService; } diff --git a/src/test/java/org/prebid/server/settings/service/DatabasePeriodicRefreshServiceTest.java b/src/test/java/org/prebid/server/settings/service/DatabasePeriodicRefreshServiceTest.java index bae78c111b4..626fffa58b7 100644 --- a/src/test/java/org/prebid/server/settings/service/DatabasePeriodicRefreshServiceTest.java +++ b/src/test/java/org/prebid/server/settings/service/DatabasePeriodicRefreshServiceTest.java @@ -2,7 +2,6 @@ import io.vertx.core.Future; import io.vertx.core.Handler; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -156,7 +155,7 @@ private void createAndInitService(long refresh) { metrics, clock); - databasePeriodicRefreshService.initialize(Promise.promise()); + databasePeriodicRefreshService.initialize(); } @SuppressWarnings("unchecked") diff --git a/src/test/java/org/prebid/server/settings/service/HttpPeriodicRefreshServiceTest.java b/src/test/java/org/prebid/server/settings/service/HttpPeriodicRefreshServiceTest.java index 3cb36048ffb..0cd1643fe2b 100644 --- a/src/test/java/org/prebid/server/settings/service/HttpPeriodicRefreshServiceTest.java +++ b/src/test/java/org/prebid/server/settings/service/HttpPeriodicRefreshServiceTest.java @@ -3,7 +3,6 @@ import com.fasterxml.jackson.core.JsonProcessingException; import io.vertx.core.Future; import io.vertx.core.Handler; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; @@ -174,7 +173,7 @@ private static void createAndInitService(CacheNotificationListener notif final HttpPeriodicRefreshService httpPeriodicRefreshService = new HttpPeriodicRefreshService( url, refreshPeriod, timeout, notificationListener, vertx, httpClient, jacksonMapper); - httpPeriodicRefreshService.initialize(Promise.promise()); + httpPeriodicRefreshService.initialize(); } @SuppressWarnings("unchecked") diff --git a/src/test/java/org/prebid/server/settings/service/S3PeriodicRefreshServiceTest.java b/src/test/java/org/prebid/server/settings/service/S3PeriodicRefreshServiceTest.java index b340b7f0ca4..34a04bf971e 100644 --- a/src/test/java/org/prebid/server/settings/service/S3PeriodicRefreshServiceTest.java +++ b/src/test/java/org/prebid/server/settings/service/S3PeriodicRefreshServiceTest.java @@ -1,7 +1,6 @@ package org.prebid.server.settings.service; import io.vertx.core.Future; -import io.vertx.core.Promise; import io.vertx.core.Vertx; import io.vertx.junit5.VertxExtension; import io.vertx.junit5.VertxTestContext; @@ -169,8 +168,6 @@ private Future createAndInitService(long refreshPeriod) { metrics, vertx); - final Promise init = Promise.promise(); - s3PeriodicRefreshService.initialize(init); - return init.future(); + return s3PeriodicRefreshService.initialize(); } }