Add autoupdate in RemoteFileSyncer (prebid#478)

Author: DGarbar

docs/config-app.md

@@ -40,10 +40,12 @@ Remote File Syncer can be related to particular entity like geolocation maxmind
 Removes and downloads file again if depending service cant process probably corrupted file in the first start.
 
 - `<SERVICE>.remote-file-syncer.download-url` - url to database file to download.
-- `<SERVICE>.remote-file-syncer.save-filepath` - local path to downloaded database file.
+- `<SERVICE>.remote-file-syncer.save-filepath` - full path to the usable file, which will be consumed by internal service.
+- `<SERVICE>.remote-file-syncer.tmp-filepath` - full path to the temporary file.
 - `<SERVICE>.remote-file-syncer.retry-count` - how many times try to download.
 - `<SERVICE>.remote-file-syncer.retry-interval-ms` - how long to wait between failed retries.
 - `<SERVICE>.remote-file-syncer.timeout-ms` - default operation timeout for obtaining database file.
+- `<SERVICE>.remote-file-syncer.update-interval-ms` - time interval between updates of the usable file.
 - `<SERVICE>.remote-file-syncer.http-client.connect-timeout-ms` - set the connect timeout.
 - `<SERVICE>.remote-file-syncer.http-client.max-redirects` - set the maximum amount of HTTP redirections to follow. A value of 0 (the default) prevents redirections from being followed.
 

src/main/java/org/prebid/server/execution/RemoteFileSyncer.java

@@ -4,16 +4,18 @@
 import io.vertx.core.Future;
 import io.vertx.core.Vertx;
 import io.vertx.core.file.AsyncFile;
+import io.vertx.core.file.CopyOptions;
 import io.vertx.core.file.FileProps;
 import io.vertx.core.file.FileSystem;
 import io.vertx.core.file.FileSystemException;
 import io.vertx.core.file.OpenOptions;
 import io.vertx.core.http.HttpClient;
-import io.vertx.core.http.HttpClientRequest;
 import io.vertx.core.http.HttpClientResponse;
+import io.vertx.core.http.HttpHeaders;
 import io.vertx.core.logging.Logger;
 import io.vertx.core.logging.LoggerFactory;
 import io.vertx.core.streams.Pump;
+import org.apache.commons.lang3.StringUtils;
 import org.prebid.server.exception.PreBidException;
 import org.prebid.server.util.HttpUtil;
 
@@ -32,38 +34,45 @@ public class RemoteFileSyncer {
 
     private final String downloadUrl;  // url to resource to be downloaded
     private final String saveFilePath; // full path on file system where downloaded file located
+    private final String tmpFilePath; // full path on file system where tmp file located
     private final int retryCount; // how many times try to download
     private final long retryInterval; // how long to wait between failed retries
     private final long timeout;
+    private final long updatePeriod;
     private final HttpClient httpClient;
     private final Vertx vertx;
     private final FileSystem fileSystem;
 
-    private RemoteFileSyncer(String downloadUrl, String saveFilePath, int retryCount,
-                             long retryInterval, long timeout, HttpClient httpClient, Vertx vertx,
+    private RemoteFileSyncer(String downloadUrl, String saveFilePath, String tmpFilePath, int retryCount,
+                             long retryInterval, long timeout, long updatePeriod, HttpClient httpClient, Vertx vertx,
                              FileSystem fileSystem) {
         this.downloadUrl = downloadUrl;
         this.saveFilePath = saveFilePath;
+        this.tmpFilePath = tmpFilePath;
         this.retryCount = retryCount;
         this.retryInterval = retryInterval;
         this.timeout = timeout;
+        this.updatePeriod = updatePeriod;
         this.httpClient = httpClient;
         this.vertx = vertx;
         this.fileSystem = fileSystem;
     }
 
-    public static RemoteFileSyncer create(String downloadUrl, String saveFilePath, int retryCount, long retryInterval,
-                                          long timeout, HttpClient httpClient, Vertx vertx) {
+    public static RemoteFileSyncer create(String downloadUrl, String saveFilePath, String tmpFilePath, int retryCount,
+                                          long retryInterval, long timeout, long updatePeriod, HttpClient httpClient,
+                                          Vertx vertx, FileSystem fileSystem) {
         HttpUtil.validateUrl(downloadUrl);
         Objects.requireNonNull(saveFilePath);
+        Objects.requireNonNull(tmpFilePath);
         Objects.requireNonNull(vertx);
         Objects.requireNonNull(httpClient);
-        final FileSystem fileSystem = vertx.fileSystem();
+        Objects.requireNonNull(fileSystem);
 
         createAndCheckWritePermissionsFor(fileSystem, saveFilePath);
+        createAndCheckWritePermissionsFor(fileSystem, tmpFilePath);
 
-        return new RemoteFileSyncer(downloadUrl, saveFilePath, retryCount, retryInterval, timeout, httpClient, vertx,
-                fileSystem);
+        return new RemoteFileSyncer(downloadUrl, saveFilePath, tmpFilePath, retryCount, retryInterval, timeout,
+                updatePeriod, httpClient, vertx, fileSystem);
     }
 
     /**
@@ -90,36 +99,68 @@ public void syncForFilepath(RemoteFileProcessor remoteFileProcessor) {
         downloadIfNotExist(remoteFileProcessor).setHandler(syncResult -> handleSync(remoteFileProcessor, syncResult));
     }
 
-    private Future<Void> downloadIfNotExist(RemoteFileProcessor fileProcessor) {
-        final Future<Void> future = Future.future();
-        fileSystem.exists(saveFilePath, existResult -> handleFileExisting(future, existResult, fileProcessor));
+    private Future<Boolean> downloadIfNotExist(RemoteFileProcessor fileProcessor) {
+        final Future<Boolean> future = Future.future();
+        checkFileExist(saveFilePath).setHandler(existResult ->
+                handleFileExistingWithSync(existResult, fileProcessor, future));
         return future;
     }
 
-    private void handleFileExisting(Future<Void> future, AsyncResult<Boolean> existResult,
-                                    RemoteFileProcessor fileProcessor) {
+    private Future<Boolean> checkFileExist(String filePath) {
+        final Future<Boolean> result = Future.future();
+        fileSystem.exists(filePath, async -> {
+            if (async.succeeded()) {
+                result.complete(async.result());
+            } else {
+                result.fail(String.format("Cant check if file exists %s", filePath));
+            }
+        });
+        return result;
+    }
+
+    private void handleFileExistingWithSync(AsyncResult<Boolean> existResult, RemoteFileProcessor fileProcessor,
+                                            Future<Boolean> future) {
         if (existResult.succeeded()) {
             if (existResult.result()) {
                 fileProcessor.setDataPath(saveFilePath)
                         .setHandler(serviceRespond -> handleServiceRespond(serviceRespond, future));
             } else {
-                tryDownload(future);
+                syncRemoteFiles().setHandler(future);
             }
         } else {
             future.fail(existResult.cause());
         }
     }
 
-    private void handleServiceRespond(AsyncResult<?> processResult, Future<Void> future) {
+    private void handleServiceRespond(AsyncResult<?> processResult, Future<Boolean> future) {
         if (processResult.failed()) {
             final Throwable cause = processResult.cause();
-            cleanUp().setHandler(removalResult -> handleCorruptedFileRemoval(removalResult, future, cause));
+            cleanUp(saveFilePath).setHandler(removalResult -> handleCorruptedFileRemoval(removalResult, future, cause));
         } else {
+            future.complete(false);
             logger.info("Existing file {0} was successfully reused for service", saveFilePath);
         }
     }
 
-    private void handleCorruptedFileRemoval(AsyncResult<Void> removalResult, Future<Void> future,
+    private Future<Void> cleanUp(String filePath) {
+        final Future<Void> future = Future.future();
+        checkFileExist(filePath).setHandler(existResult -> handleFileExistsWithDelete(filePath, existResult, future));
+        return future;
+    }
+
+    private void handleFileExistsWithDelete(String filePath, AsyncResult<Boolean> existResult, Future<Void> future) {
+        if (existResult.succeeded()) {
+            if (existResult.result()) {
+                fileSystem.delete(filePath, future);
+            } else {
+                future.complete();
+            }
+        } else {
+            future.fail(new PreBidException(String.format("Cant check if file exists %s", filePath)));
+        }
+    }
+
+    private void handleCorruptedFileRemoval(AsyncResult<Void> removalResult, Future<Boolean> future,
                                             Throwable serviceCause) {
         if (removalResult.failed()) {
             final Throwable cause = removalResult.cause();
@@ -130,29 +171,42 @@ private void handleCorruptedFileRemoval(AsyncResult<Void> removalResult, Future<
             logger.info("Existing file {0} cant be processed by service, try to download after removal",
                     serviceCause, saveFilePath);
 
-            tryDownload(future);
+            syncRemoteFiles().setHandler(future);
         }
     }
 
-    private void tryDownload(Future<Void> future) {
-        download().setHandler(downloadResult -> handleDownload(future, downloadResult));
+    private Future<Boolean> syncRemoteFiles() {
+        return tryDownload()
+                .compose(downloadResult -> swapFiles())
+                .map(true);
+    }
+
+    private Future<Void> tryDownload() {
+        final Future<Void> result = Future.future();
+        cleanUp(tmpFilePath).setHandler(event -> handleTmpDelete(event, result));
+        return result;
+    }
+
+    private void handleTmpDelete(AsyncResult<Void> tmpDeleteResult, Future<Void> result) {
+        if (tmpDeleteResult.failed()) {
+            result.fail(tmpDeleteResult.cause());
+        } else {
+            download().setHandler(downloadResult -> handleDownload(downloadResult, result));
+        }
     }
 
     private Future<Void> download() {
         final Future<Void> future = Future.future();
         final OpenOptions openOptions = new OpenOptions().setCreateNew(true);
-        fileSystem.open(saveFilePath, openOptions, openResult -> handleFileOpenWithDownload(future, openResult));
+        fileSystem.open(tmpFilePath, openOptions, openResult -> handleFileOpenWithDownload(openResult, future));
         return future;
     }
 
-    private void handleFileOpenWithDownload(Future<Void> future, AsyncResult<AsyncFile> openResult) {
+    private void handleFileOpenWithDownload(AsyncResult<AsyncFile> openResult, Future<Void> future) {
         if (openResult.succeeded()) {
             final AsyncFile asyncFile = openResult.result();
             try {
-                // .getNow is not working
-                final HttpClientRequest httpClientRequest = httpClient
-                        .getAbs(downloadUrl, response -> pumpFileFromRequest(response, asyncFile, future));
-                httpClientRequest.end();
+                httpClient.getAbs(downloadUrl, response -> pumpFileFromRequest(response, asyncFile, future)).end();
             } catch (Exception e) {
                 future.fail(e);
             }
@@ -162,35 +216,35 @@ private void handleFileOpenWithDownload(Future<Void> future, AsyncResult<AsyncFi
     }
 
     private void pumpFileFromRequest(HttpClientResponse httpClientResponse, AsyncFile asyncFile, Future<Void> future) {
-        logger.info("Trying to download from {0}", downloadUrl);
+        logger.info("Trying to download file from {0}", downloadUrl);
         httpClientResponse.pause();
         final Pump pump = Pump.pump(httpClientResponse, asyncFile);
         pump.start();
         httpClientResponse.resume();
 
-        final long idTimer = setTimeoutTimer(asyncFile, future, pump);
+        final long idTimer = setTimeoutTimer(asyncFile, pump, future);
 
-        httpClientResponse.endHandler(responseEndResult -> handleResponseEnd(asyncFile, future, idTimer));
+        httpClientResponse.endHandler(responseEndResult -> handleResponseEnd(asyncFile, idTimer, future));
     }
 
-    private long setTimeoutTimer(AsyncFile asyncFile, Future<Void> future, Pump pump) {
-        return vertx.setTimer(timeout, timerId -> handleTimeout(asyncFile, future, pump));
+    private long setTimeoutTimer(AsyncFile asyncFile, Pump pump, Future<Void> future) {
+        return vertx.setTimer(timeout, timerId -> handleTimeout(asyncFile, pump, future));
     }
 
-    private void handleTimeout(AsyncFile asyncFile, Future<Void> future, Pump pump) {
+    private void handleTimeout(AsyncFile asyncFile, Pump pump, Future<Void> future) {
         pump.stop();
         asyncFile.close();
         if (!future.isComplete()) {
             future.fail(new TimeoutException("Timeout on download"));
         }
     }
 
-    private void handleResponseEnd(AsyncFile asyncFile, Future<Void> future, long idTimer) {
+    private void handleResponseEnd(AsyncFile asyncFile, long idTimer, Future<Void> future) {
         vertx.cancelTimer(idTimer);
         asyncFile.flush().close(future);
     }
 
-    private void handleDownload(Future<Void> future, AsyncResult<Void> downloadResult) {
+    private void handleDownload(AsyncResult<Void> downloadResult, Future<Void> future) {
         if (downloadResult.failed()) {
             retryDownload(future, retryInterval, retryCount);
         } else {
@@ -206,45 +260,47 @@ private void retryDownload(Future<Void> receivedFuture, long retryInterval, long
     private void handleRetry(Future<Void> receivedFuture, long retryInterval, long retryCount) {
         if (retryCount > 0) {
             final long next = retryCount - 1;
-            cleanUp().compose(aVoid -> download())
-                    .setHandler(retryResult -> handleRetryResult(receivedFuture, retryInterval, next, retryResult));
+            cleanUp(tmpFilePath).compose(ignore -> download())
+                    .setHandler(retryResult -> handleRetryResult(retryInterval, next, retryResult, receivedFuture));
         } else {
-            cleanUp().setHandler(aVoid -> receivedFuture.fail(new PreBidException("File sync failed after retries")));
+            cleanUp(tmpFilePath).setHandler(ignore -> receivedFuture.fail(new PreBidException(
+                    String.format("File sync failed after %s retries", this.retryCount - retryCount))));
         }
     }
 
-    private Future<Void> cleanUp() {
-        final Future<Void> future = Future.future();
-        fileSystem.exists(saveFilePath, existResult -> handleFileExistsWithDelete(future, existResult));
-        return future;
-    }
-
-    private void handleFileExistsWithDelete(Future<Void> future, AsyncResult<Boolean> existResult) {
-        if (existResult.succeeded()) {
-            if (existResult.result()) {
-                fileSystem.delete(saveFilePath, future);
-            } else {
-                future.complete();
-            }
-        } else {
-            future.fail(new PreBidException(String.format("Cant check if file exists %s", saveFilePath)));
-        }
-    }
-
-    private void handleRetryResult(Future<Void> future, long retryInterval, long next, AsyncResult<Void> retryResult) {
+    private void handleRetryResult(long retryInterval, long next, AsyncResult<Void> retryResult, Future<Void> future) {
         if (retryResult.succeeded()) {
             future.complete();
         } else {
             retryDownload(future, retryInterval, next);
         }
     }
 
-    private void handleSync(RemoteFileProcessor remoteFileProcessor, AsyncResult<Void> syncResult) {
+    private Future<Void> swapFiles() {
+        final Future<Void> result = Future.future();
+        logger.info("Sync {0} to {1}", tmpFilePath, saveFilePath);
+
+        final CopyOptions copyOptions = new CopyOptions().setReplaceExisting(true);
+        fileSystem.move(tmpFilePath, saveFilePath, copyOptions, result);
+        return result;
+    }
+
+    private void handleSync(RemoteFileProcessor remoteFileProcessor, AsyncResult<Boolean> syncResult) {
         if (syncResult.succeeded()) {
-            remoteFileProcessor.setDataPath(saveFilePath)
-                    .setHandler(this::logFileProcessStatus);
+            if (syncResult.result()) {
+                logger.info("Sync service for {0}", saveFilePath);
+                remoteFileProcessor.setDataPath(saveFilePath)
+                        .setHandler(this::logFileProcessStatus);
+            } else {
+                logger.info("Sync is not required for {0}", saveFilePath);
+            }
         } else {
-            logger.error("Cant download file from {0}", syncResult.cause(), downloadUrl);
+            logger.error("Cant sync file from {0}", syncResult.cause(), downloadUrl);
+        }
+
+        // setup new update regardless of the result
+        if (updatePeriod > 0) {
+            vertx.setTimer(updatePeriod, idUpdateNew -> configureAutoUpdates(remoteFileProcessor));
         }
     }
 
@@ -255,5 +311,45 @@ private void logFileProcessStatus(AsyncResult<?> serviceRespond) {
             logger.error("Service cant process file {0} and still unavailable.", saveFilePath);
         }
     }
+
+    private void configureAutoUpdates(RemoteFileProcessor remoteFileProcessor) {
+        logger.info("Check for updated for {0}", saveFilePath);
+        tryUpdate().setHandler(asyncUpdate -> {
+            if (asyncUpdate.failed()) {
+                logger.warn("File {0} update failed", asyncUpdate.cause(), saveFilePath);
+            }
+            handleSync(remoteFileProcessor, asyncUpdate);
+        });
+    }
+
+    private Future<Boolean> tryUpdate() {
+        return checkFileExist(saveFilePath)
+                .compose(fileExists -> fileExists ? isNeedToUpdate() : Future.succeededFuture(true))
+                .compose(needUpdate -> needUpdate ? syncRemoteFiles() : Future.succeededFuture(false));
+    }
+
+    private Future<Boolean> isNeedToUpdate() {
+        final Future<Boolean> isNeedToUpdate = Future.future();
+        httpClient.headAbs(downloadUrl, response -> checkNewVersion(response, isNeedToUpdate))
+                .exceptionHandler(isNeedToUpdate::fail)
+                .end();
+        return isNeedToUpdate;
+    }
+
+    private void checkNewVersion(HttpClientResponse response, Future<Boolean> isNeedToUpdate) {
+        final String contentLengthParameter = response.getHeader(HttpHeaders.CONTENT_LENGTH);
+        if (StringUtils.isNumeric(contentLengthParameter) && !contentLengthParameter.equals("0")) {
+            final long contentLength = Long.parseLong(contentLengthParameter);
+            fileSystem.props(saveFilePath, filePropsResult -> {
+                if (filePropsResult.succeeded()) {
+                    isNeedToUpdate.complete(filePropsResult.result().size() != contentLength);
+                } else {
+                    isNeedToUpdate.fail(filePropsResult.cause());
+                }
+            });
+        } else {
+            isNeedToUpdate.fail(String.format("ContentLength is invalid: %s", contentLengthParameter));
+        }
+    }
 }
 

src/main/java/org/prebid/server/spring/config/GeoLocationConfiguration.java

@@ -75,9 +75,10 @@ private GeoLocationService createGeoLocationService(RemoteFileSyncerProperties f
                     .setMaxRedirects(httpClientProperties.getMaxRedirects());
 
             final RemoteFileSyncer remoteFileSyncer = RemoteFileSyncer.create(fileSyncerProperties.getDownloadUrl(),
-                    fileSyncerProperties.getSaveFilepath(), fileSyncerProperties.getRetryCount(),
-                    fileSyncerProperties.getRetryIntervalMs(), fileSyncerProperties.getTimeoutMs(),
-                    vertx.createHttpClient(httpClientOptions), vertx);
+                    fileSyncerProperties.getSaveFilepath(), fileSyncerProperties.getTmpFilepath(),
+                    fileSyncerProperties.getRetryCount(), fileSyncerProperties.getRetryIntervalMs(),
+                    fileSyncerProperties.getTimeoutMs(), fileSyncerProperties.getUpdateIntervalMs(),
+                    vertx.createHttpClient(httpClientOptions), vertx, vertx.fileSystem());
             final MaxMindGeoLocationService maxMindGeoLocationService = new MaxMindGeoLocationService();
 
             remoteFileSyncer.syncForFilepath(maxMindGeoLocationService);