Split out file distribution RPC methods into its own class

6 files changed, 178 insertions, 121 deletions

diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServer.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServer.java
index ae0360fecf2..fe5976e9c1f 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServer.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServer.java
@@ -1,7 +1,6 @@
 // Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
 package com.yahoo.vespa.config.proxy;
 
-import com.yahoo.config.FileReference;
 import com.yahoo.jrt.*;
 import com.yahoo.log.LogLevel;
 import com.yahoo.vespa.config.*;
@@ -10,16 +9,10 @@ import com.yahoo.vespa.config.protocol.JRTConfigRequestFactory;
 import com.yahoo.vespa.config.protocol.JRTServerConfigRequest;
 import com.yahoo.vespa.config.protocol.JRTServerConfigRequestV3;
 
-import java.io.File;
 import java.util.Arrays;
 import java.util.Iterator;
-import java.util.List;
-import java.util.Map;
-import java.util.Optional;
 import java.util.Set;
 import java.util.logging.Logger;
-import java.util.stream.Collectors;
-import java.util.stream.Stream;
 
 
 /**
@@ -34,14 +27,14 @@ public class ConfigProxyRpcServer implements Runnable, TargetWatcher, RpcServer
     private static final int TRACELEVEL = 6;
 
     private final Spec spec;
-    private final Supervisor supervisor = new Supervisor(new Transport());
+    private final Supervisor supervisor;
     private final ProxyServer proxyServer;
 
-    ConfigProxyRpcServer(ProxyServer proxyServer, Spec spec) {
+    ConfigProxyRpcServer(ProxyServer proxyServer, Supervisor supervisor, Spec spec) {
         this.proxyServer = proxyServer;
         this.spec = spec;
+        this.supervisor = supervisor;
         declareConfigMethods();
-        declareFileDistributionMethods();
     }
 
     public void run() {
@@ -109,40 +102,6 @@ public class ConfigProxyRpcServer implements Runnable, TargetWatcher, RpcServer
                 .returnDesc(0, "ret", "Empty string or error message"));
     }
 
-    private void declareFileDistributionMethods() {
-        // Legacy method, needs to be the same name as used in filedistributor
-        supervisor.addMethod(new Method("waitFor", "s", "s",
-                this, "getFile")
-                .methodDesc("get path to file reference")
-                .paramDesc(0, "file reference", "file reference")
-                .returnDesc(0, "path", "path to file"));
-        supervisor.addMethod(new Method("filedistribution.getFile", "s", "s",
-                this, "getFile")
-                .methodDesc("get path to file reference")
-                .paramDesc(0, "file reference", "file reference")
-                .returnDesc(0, "path", "path to file"));
-        supervisor.addMethod(new Method("filedistribution.getActiveFileReferencesStatus", "", "SD",
-                this, "getActiveFileReferencesStatus")
-                .methodDesc("download status for file references")
-                .returnDesc(0, "file references", "array of file references")
-                .returnDesc(1, "download status", "percentage downloaded of each file reference in above array"));
-        supervisor.addMethod(new Method("filedistribution.setFileReferencesToDownload", "S", "i",
-                this, "setFileReferencesToDownload")
-                .methodDesc("set which file references to download")
-                .paramDesc(0, "file references", "file reference to download")
-                .returnDesc(0, "ret", "0 if success, 1 otherwise"));
-        supervisor.addMethod(new Method("filedistribution.receiveFile", "ssxlis", "i", // TODO Temporary method to get started with testing
-                this, "receiveFile")
-                .methodDesc("receive file reference content")
-                .paramDesc(0, "file references", "file reference to download")
-                .paramDesc(1, "filename", "filename")
-                .paramDesc(2, "content", "array of bytes")
-                .paramDesc(3, "hash", "xx64hash of the file content")
-                .paramDesc(4, "errorcode", "Error code. 0 if none")
-                .paramDesc(5, "error-description", "Error description.")
-                .returnDesc(0, "ret", "0 if success, 1 otherwise"));
-    }
-
     //---------------- RPC methods ------------------------------------
 
     /**
@@ -249,75 +208,6 @@ public class ConfigProxyRpcServer implements Runnable, TargetWatcher, RpcServer
         req.returnValues().add(new StringValue(memoryCache.dumpCacheToDisk(req.parameters().get(0).asString(), memoryCache)));
     }
 
-    // TODO: Duplicate of code in FileAcquirereImpl. Find out where to put it. What about C++ code using this RPC call?
-    private static final int baseErrorCode = 0x10000;
-    private static final int baseFileProviderErrorCode = baseErrorCode + 0x1000;
-
-    private static final int fileReferenceDoesNotExists = baseFileProviderErrorCode;
-    private static final int fileReferenceRemoved = fileReferenceDoesNotExists + 1;
-    private static final int fileReferenceInternalError = fileReferenceRemoved + 1;
-
-    @SuppressWarnings({"UnusedDeclaration"})
-    public final void getFile(Request req) {
-        req.detach();
-        FileReference fileReference = new FileReference(req.parameters().get(0).asString());
-        log.log(LogLevel.DEBUG, "getFile() called for file reference '" + fileReference.value() + "'");
-        Optional<File> pathToFile = proxyServer.fileDownloader().getFile(fileReference);
-        try {
-            if (pathToFile.isPresent()) {
-                req.returnValues().add(new StringValue(pathToFile.get().getAbsolutePath()));
-                log.log(LogLevel.INFO, "File reference '" + fileReference.value() + "' available at " + pathToFile.get());
-            } else {
-                log.log(LogLevel.INFO, "File reference '" + fileReference.value() + "' not found, returning error");
-                req.setError(fileReferenceDoesNotExists, "File reference '" + fileReference.value() + "' not found");
-            }
-        } catch (Throwable e) {
-            log.log(LogLevel.WARNING, "File reference '" + fileReference.value() + "' got exeption: " + e.getMessage());
-            req.setError(fileReferenceInternalError, "File reference '" + fileReference.value() + "' removed");
-        }
-        req.returnRequest();
-    }
-
-    @SuppressWarnings({"UnusedDeclaration"})
-    public final void getActiveFileReferencesStatus(Request req) {
-        Map<FileReference, Double> downloadStatus = proxyServer.fileDownloader().downloadStatus();
-
-        String[] fileRefArray = new String[downloadStatus.keySet().size()];
-        fileRefArray = downloadStatus.keySet().stream()
-                .map(FileReference::value)
-                .collect(Collectors.toList())
-                .toArray(fileRefArray);
-
-        double[] downloadStatusArray = new double[downloadStatus.values().size()];
-        int i = 0;
-        for (Double d : downloadStatus.values()) {
-            downloadStatusArray[i++] = d;
-        }
-
-        req.returnValues().add(new StringArray(fileRefArray));
-        req.returnValues().add(new DoubleArray(downloadStatusArray));
-    }
-
-    @SuppressWarnings({"UnusedDeclaration"})
-    public final void setFileReferencesToDownload(Request req) {
-        String[] fileReferenceStrings = req.parameters().get(0).asStringArray();
-        List<FileReference> fileReferences = Stream.of(fileReferenceStrings)
-                .map(FileReference::new)
-                .collect(Collectors.toList());
-        proxyServer.fileDownloader().queueForDownload(fileReferences);
-
-        req.returnValues().add(new Int32Value(0));
-    }
-
-    @SuppressWarnings({"UnusedDeclaration"})
-    public final void receiveFile(Request req) {
-        FileReference fileReference = new FileReference(req.parameters().get(0).asString());
-        String filename = req.parameters().get(1).asString();
-        byte[] content = req.parameters().get(2).asData();
-        proxyServer.fileDownloader().receiveFile(fileReference, filename, content);
-        req.returnValues().add(new Int32Value(0));
-    }
-
     //----------------------------------------------------
 
     private boolean isProtocolVersionSupported(JRTServerConfigRequest request) {
diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ProxyServer.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ProxyServer.java
index 5668852311f..173d2b8a43a 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ProxyServer.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/ProxyServer.java
@@ -5,12 +5,15 @@ import com.yahoo.concurrent.DaemonThreadFactory;
 import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.jrt.Spec;
 
+import com.yahoo.jrt.Supervisor;
+import com.yahoo.jrt.Transport;
 import com.yahoo.log.LogLevel;
 import com.yahoo.log.LogSetup;
 import com.yahoo.log.event.Event;
 import com.yahoo.system.CatchSigTerm;
 import com.yahoo.vespa.config.*;
 import com.yahoo.vespa.config.protocol.JRTServerConfigRequest;
+import com.yahoo.vespa.config.proxy.filedistribution.FileDistributionRpcServer;
 import com.yahoo.vespa.config.proxy.filedistribution.FileDownloader;
 
 import java.util.List;
@@ -40,6 +43,7 @@ public class ProxyServer implements Runnable {
 
     // Scheduled executor that periodically checks for requests that have timed out and response should be returned to clients
     private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1, new DaemonThreadFactory());
+    private final Supervisor supervisor = new Supervisor(new Transport());
     private final ClientUpdater clientUpdater;
     private ScheduledFuture<?> delayedResponseScheduler;
 
@@ -84,6 +88,7 @@ public class ProxyServer implements Runnable {
         clientUpdater = new ClientUpdater(rpcServer, statistics, delayedResponses);
         this.configClient = createClient(clientUpdater, delayedResponses, source, timingValues, memoryCache, configClient);
         this.fileDownloader = new FileDownloader(new JRTConnectionPool(source));
+        new FileDistributionRpcServer(supervisor, fileDownloader);
     }
 
     static ProxyServer createTestServer(ConfigSourceSet source) {
@@ -162,7 +167,7 @@ public class ProxyServer implements Runnable {
     }
 
     private ConfigProxyRpcServer createRpcServer(Spec spec) {
-        return  (spec == null) ? null : new ConfigProxyRpcServer(this, spec); // TODO: Try to avoid first argument being 'this'
+        return  (spec == null) ? null : new ConfigProxyRpcServer(this, supervisor, spec); // TODO: Try to avoid first argument being 'this'
     }
 
     private RpcConfigSourceClient createRpcClient() {
diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionRpcServer.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionRpcServer.java
new file mode 100644
index 00000000000..552758bcfa4
--- /dev/null
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileDistributionRpcServer.java
@@ -0,0 +1,153 @@
+// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
+package com.yahoo.vespa.config.proxy.filedistribution;
+
+import com.yahoo.config.FileReference;
+import com.yahoo.jrt.DoubleArray;
+import com.yahoo.jrt.Int32Value;
+import com.yahoo.jrt.Method;
+import com.yahoo.jrt.Request;
+import com.yahoo.jrt.StringArray;
+import com.yahoo.jrt.StringValue;
+import com.yahoo.jrt.Supervisor;
+import com.yahoo.log.LogLevel;
+
+import java.io.File;
+import java.util.List;
+import java.util.Map;
+import java.util.Optional;
+import java.util.logging.Logger;
+import java.util.stream.Collectors;
+import java.util.stream.Stream;
+
+/**
+ * An RPC server that handles file distribution requests.
+ *
+ * @author hmusum
+ */
+public class FileDistributionRpcServer {
+
+    private final static Logger log = Logger.getLogger(FileDistributionRpcServer.class.getName());
+
+    private final Supervisor supervisor;
+    private final FileDownloader downloader;
+
+    public FileDistributionRpcServer(Supervisor supervisor, FileDownloader downloader) {
+        this.supervisor = supervisor;
+        this.downloader = downloader;
+        declareFileDistributionMethods();
+    }
+
+    private void declareFileDistributionMethods() {
+        // Legacy method, needs to be the same name as used in filedistributor
+        supervisor.addMethod(new Method("waitFor", "s", "s",
+                                        this, "getFile")
+                                     .methodDesc("get path to file reference")
+                                     .paramDesc(0, "file reference", "file reference")
+                                     .returnDesc(0, "path", "path to file"));
+        supervisor.addMethod(new Method("filedistribution.getFile", "s", "s",
+                                        this, "getFile")
+                                     .methodDesc("get path to file reference")
+                                     .paramDesc(0, "file reference", "file reference")
+                                     .returnDesc(0, "path", "path to file"));
+        supervisor.addMethod(new Method("filedistribution.getActiveFileReferencesStatus", "", "SD",
+                                        this, "getActiveFileReferencesStatus")
+                                     .methodDesc("download status for file references")
+                                     .returnDesc(0, "file references", "array of file references")
+                                     .returnDesc(1, "download status", "percentage downloaded of each file reference in above array"));
+        supervisor.addMethod(new Method("filedistribution.setFileReferencesToDownload", "S", "i",
+                                        this, "setFileReferencesToDownload")
+                                     .methodDesc("set which file references to download")
+                                     .paramDesc(0, "file references", "file reference to download")
+                                     .returnDesc(0, "ret", "0 if success, 1 otherwise"));
+        supervisor.addMethod(new Method("filedistribution.receiveFile", "ssxlis", "i", // TODO Temporary method to get started with testing
+                                        this, "receiveFile")
+                                     .methodDesc("receive file reference content")
+                                     .paramDesc(0, "file references", "file reference to download")
+                                     .paramDesc(1, "filename", "filename")
+                                     .paramDesc(2, "content", "array of bytes")
+                                     .paramDesc(3, "hash", "xx64hash of the file content")
+                                     .paramDesc(4, "errorcode", "Error code. 0 if none")
+                                     .paramDesc(5, "error-description", "Error description.")
+                                     .returnDesc(0, "ret", "0 if success, 1 otherwise"));
+    }
+
+    //---------------- RPC methods ------------------------------------
+    // TODO: Duplicate of code in FileAcquirereImpl. Find out where to put it. What about C++ code using this RPC call?
+    private static final int baseErrorCode = 0x10000;
+    private static final int baseFileProviderErrorCode = baseErrorCode + 0x1000;
+
+    private static final int fileReferenceDoesNotExists = baseFileProviderErrorCode;
+    private static final int fileReferenceRemoved = fileReferenceDoesNotExists + 1;
+    private static final int fileReferenceInternalError = fileReferenceRemoved + 1;
+
+    @SuppressWarnings({"UnusedDeclaration"})
+    public final void getFile(Request req) {
+        req.detach();
+        FileReference fileReference = new FileReference(req.parameters().get(0).asString());
+        log.log(LogLevel.DEBUG, "getFile() called for file reference '" + fileReference.value() + "'");
+        Optional<File> pathToFile = downloader.getFile(fileReference);
+        try {
+            if (pathToFile.isPresent()) {
+                req.returnValues().add(new StringValue(pathToFile.get().getAbsolutePath()));
+                log.log(LogLevel.INFO, "File reference '" + fileReference.value() + "' available at " + pathToFile.get());
+            } else {
+                log.log(LogLevel.INFO, "File reference '" + fileReference.value() + "' not found, returning error");
+                req.setError(fileReferenceDoesNotExists, "File reference '" + fileReference.value() + "' not found");
+            }
+        } catch (Throwable e) {
+            log.log(LogLevel.WARNING, "File reference '" + fileReference.value() + "' got exeption: " + e.getMessage());
+            req.setError(fileReferenceInternalError, "File reference '" + fileReference.value() + "' removed");
+        }
+        req.returnRequest();
+    }
+
+    @SuppressWarnings({"UnusedDeclaration"})
+    public final void getActiveFileReferencesStatus(Request req) {
+        Map<FileReference, Double> downloadStatus = downloader.downloadStatus();
+
+        String[] fileRefArray = new String[downloadStatus.keySet().size()];
+        fileRefArray = downloadStatus.keySet().stream()
+                .map(FileReference::value)
+                .collect(Collectors.toList())
+                .toArray(fileRefArray);
+
+        double[] downloadStatusArray = new double[downloadStatus.values().size()];
+        int i = 0;
+        for (Double d : downloadStatus.values()) {
+            downloadStatusArray[i++] = d;
+        }
+
+        req.returnValues().add(new StringArray(fileRefArray));
+        req.returnValues().add(new DoubleArray(downloadStatusArray));
+    }
+
+    @SuppressWarnings({"UnusedDeclaration"})
+    public final void setFileReferencesToDownload(Request req) {
+        String[] fileReferenceStrings = req.parameters().get(0).asStringArray();
+        List<FileReference> fileReferences = Stream.of(fileReferenceStrings)
+                .map(FileReference::new)
+                .collect(Collectors.toList());
+        downloader.queueForDownload(fileReferences);
+
+        req.returnValues().add(new Int32Value(0));
+    }
+
+    @SuppressWarnings({"UnusedDeclaration"})
+    public final void receiveFile(Request req) {
+        FileReference fileReference = new FileReference(req.parameters().get(0).asString());
+        String filename = req.parameters().get(1).asString();
+        byte[] content = req.parameters().get(2).asData();
+        long xxhash = req.parameters().get(3).asInt64();
+        int errorCode = req.parameters().get(3).asInt32();
+        String errorDescription = req.parameters().get(4).asString();
+
+        if (errorCode == 0) {
+            //downloader.receive(fileReference, filename, content);
+            req.returnValues().add(new Int32Value(0));
+        } else {
+            log.log(LogLevel.WARNING, "Receiving file reference '" + fileReference.value() + "' failed: " + errorDescription);
+            req.returnValues().add(new Int32Value(1));
+            // TODO: Add error description return value here too?
+        }
+    }
+}
diff --git a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileReferenceDownloader.java b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileReferenceDownloader.java
index 917374740f1..611ad67a5d8 100644
--- a/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileReferenceDownloader.java
+++ b/config-proxy/src/main/java/com/yahoo/vespa/config/proxy/filedistribution/FileReferenceDownloader.java
@@ -162,7 +162,7 @@ class FileReferenceDownloader {
             return false;
         } else if (request.returnValues().size() == 0) {
             return false;
-        } else if (!request.checkReturnTypes("i")) {
+        } else if (!request.checkReturnTypes("is")) { // TODO: Do not hard-code return type
             log.log(LogLevel.WARNING, "Invalid return types for response: " + request.errorMessage());
             return false;
         }
@@ -180,4 +180,5 @@ class FileReferenceDownloader {
     Map<FileReference, Double> downloadStatus() {
         return ImmutableMap.copyOf(downloadStatus);
     }
+
 }
diff --git a/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServerTest.java b/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServerTest.java
index f9b334a6f87..4a9d2acb4c5 100644
--- a/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServerTest.java
+++ b/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/ConfigProxyRpcServerTest.java
@@ -5,6 +5,8 @@ import com.yahoo.config.subscription.ConfigSourceSet;
 import com.yahoo.jrt.Request;
 import com.yahoo.jrt.Spec;
 import com.yahoo.jrt.StringValue;
+import com.yahoo.jrt.Supervisor;
+import com.yahoo.jrt.Transport;
 import com.yahoo.vespa.config.RawConfig;
 import org.junit.After;
 import org.junit.Before;
@@ -28,7 +30,7 @@ public class ConfigProxyRpcServerTest {
     @Before
     public void setup() {
         proxyServer = ProxyServer.createTestServer(new ConfigSourceSet(address));
-        rpcServer = new ConfigProxyRpcServer(proxyServer, null);
+        rpcServer = new ConfigProxyRpcServer(proxyServer, new Supervisor(new Transport()), null);
     }
 
     @After
@@ -40,7 +42,7 @@ public class ConfigProxyRpcServerTest {
     public void basic() {
         ProxyServer proxy = ProxyServer.createTestServer(new MockConfigSource(new MockClientUpdater()));
         Spec spec = new Spec("localhost", 12345);
-        ConfigProxyRpcServer server = new ConfigProxyRpcServer(proxy, spec);
+        ConfigProxyRpcServer server = new ConfigProxyRpcServer(proxy, new Supervisor(new Transport()), spec);
         assertThat(server.getSpec(), is(spec));
     }
 
diff --git a/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/filedistribution/FileDownloaderTest.java b/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/filedistribution/FileDownloaderTest.java
index 18d49e9a224..c44e19f9f03 100644
--- a/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/filedistribution/FileDownloaderTest.java
+++ b/config-proxy/src/test/java/com/yahoo/vespa/config/proxy/filedistribution/FileDownloaderTest.java
@@ -5,6 +5,9 @@ import com.yahoo.io.IOUtils;
 import com.yahoo.jrt.Int32Value;
 import com.yahoo.jrt.Request;
 import com.yahoo.jrt.RequestWaiter;
+import com.yahoo.jrt.StringValue;
+import com.yahoo.jrt.Supervisor;
+import com.yahoo.jrt.Transport;
 import com.yahoo.text.Utf8;
 import com.yahoo.vespa.config.Connection;
 import com.yahoo.vespa.config.ConnectionPool;
@@ -18,9 +21,7 @@ import java.time.Duration;
 import java.util.Arrays;
 import java.util.LinkedHashSet;
 import java.util.List;
-import java.util.Map;
 import java.util.Optional;
-import java.util.Set;
 
 import static org.junit.Assert.assertEquals;
 import static org.junit.Assert.assertFalse;
@@ -38,6 +39,7 @@ public class FileDownloaderTest {
             File downloadDir = Files.createTempDirectory("filedistribution").toFile();
             connection = new MockConnection();
             fileDownloader = new FileDownloader(connection, downloadDir, Duration.ofMillis(3000));
+            FileDistributionRpcServer rpcServer = new FileDistributionRpcServer(new Supervisor(new Transport()), fileDownloader);
         } catch (IOException e) {
             e.printStackTrace();
             fail(e.getMessage());
@@ -204,8 +206,10 @@ public class FileDownloaderTest {
 
             @Override
             public void request(Request request) {
-                if (request.methodName().equals("filedistribution.serveFile"))
+                if (request.methodName().equals("filedistribution.serveFile")) {
                     request.returnValues().add(new Int32Value(0));
+                    request.returnValues().add(new StringValue("OK"));
+                }
             }
         }
 
@@ -213,8 +217,10 @@ public class FileDownloaderTest {
 
             @Override
             public void request(Request request) {
-                if (request.methodName().equals("filedistribution.serveFile"))
+                if (request.methodName().equals("filedistribution.serveFile")) {
                     request.returnValues().add(new Int32Value(1));
+                    request.returnValues().add(new StringValue("Internal error"));
+                }
             }
         }