// Copyright 2016 Yahoo Inc. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.filedistribution;

import java.io.File;
import java.io.IOException;
import java.util.Collection;
import java.util.List;
import java.util.concurrent.locks.Lock;

/**
 * @author tonytv
 */
public class FileDistributionManager {

    private final static boolean available;

    @SuppressWarnings({"UnusedDeclaration"}) // Needs to be here due to JNI
    private long nativeFileDistributionManager;

    private final File applicationDirectory;
    private final String appId;
    private final Lock lock;


    private native static void setup();

    private native void init(byte[] fileDbDirectory, byte[] zkServers);

    private native String addFileImpl(byte[] absolutePath);

    private native void setDeployedFilesImpl(byte[] host, byte[] appId, byte[][] fileReferences);

    private native void limitSendingOfDeployedFilesToImpl(byte[][] hostNames, byte[] appId);
    private native void limitFilesToImpl(byte[][] fileReferences);
    private native void removeDeploymentsThatHaveDifferentApplicationIdImpl(byte[][] asByteArrays, byte[] bytes);

    private native byte[] getProgressImpl(byte[] fileReference,
                                          byte[][] hostNames);

    private byte[][] getAsByteArrays(Collection<String> strings) {
        byte[][] byteArrays = new byte[strings.size()][];
        int i = 0;
        for (String string : strings) {
            byteArrays[i++] = string.getBytes();
        }
        return byteArrays;
    }

    @Override
    protected void finalize() throws Throwable {
        shutdown();
        super.finalize();
    }

    static {
        available = loadLibrary("filedistributionmanager");
        if (available)
            setup();
    }

    private static boolean loadLibrary(String name) {
        try {
            System.loadLibrary(name);
            return true;
        }
        catch (UnsatisfiedLinkError e) {
            return false;
        }
    }

    /** Returns whether this functionality is available in this runtime */
    public static boolean isAvailable() { return available; }

    private byte[] absolutePath(File file) {
        return file.getAbsolutePath().getBytes();
    }

    private void ensureDirExists(File dir) {
        if (!dir.exists()) {
            throw new PathDoesNotExistException(dir.getPath());
        }
    }

    public FileDistributionManager(File fileDbDirectory, File applicationDirectory, String zkServers, String appId, Lock lock) {
        ensureDirExists(applicationDirectory);
        ensureDirExists(fileDbDirectory);

        this.applicationDirectory = applicationDirectory;
        this.appId = appId;
        this.lock = lock;

        init(fileDbDirectory.getPath().getBytes(), zkServers.getBytes());
    }

    public String addFile(String relativePath) {
        File path = new File(applicationDirectory, relativePath);
        if (!path.exists())
            throw new PathDoesNotExistException(path.getPath());

        try (LockGuard guard = new LockGuard(lock)) {
            return addFileImpl(absolutePath(path));
        }
    }

    public void setDeployedFiles(String hostName, Collection<String> fileReferences) {
        try (LockGuard guard = new LockGuard(lock)) {
            setDeployedFilesImpl(hostName.getBytes(), appId.getBytes(), getAsByteArrays(fileReferences));
        }
    }

    public void reloadDeployFileDistributor() {
        try (LockGuard guard = new LockGuard(lock)) {
            Runtime.getRuntime().exec("pkill -SIGUSR1 -x filedistributor");
        } catch (IOException e) {
            throw new RuntimeException("Failed to reinitialize the filedistributor", e);
        }
    }

    public void limitSendingOfDeployedFilesTo(Collection<String> hostNames) {
        try (LockGuard guard = new LockGuard(lock)) {
            limitSendingOfDeployedFilesToImpl(getAsByteArrays(hostNames), appId.getBytes());
        }
    }

    public byte[] getProgress(String fileReference,
                              List<String> hostNamesSortedAscending) {
        return getProgressImpl(fileReference.getBytes(), getAsByteArrays(hostNamesSortedAscending));
    }


    public native void shutdown();

    public void removeDeploymentsThatHaveDifferentApplicationId(Collection<String> targetHostnames) {
        try (LockGuard guard = new LockGuard(lock)) {
            removeDeploymentsThatHaveDifferentApplicationIdImpl(getAsByteArrays(targetHostnames), appId.getBytes());
        }
    }

    private static class LockGuard implements AutoCloseable {
        private final Lock lock;
        public LockGuard(Lock lock) {
            this.lock = lock;
            lock.lock();
        }

        @Override
        public void close() {
            lock.unlock();
        }
    }
}

@@ -150,16 +150,117 @@
 ution;%0A%0A
+import com.yahoo.system.execution.ProcessExecutor;%0Aimport com.yahoo.system.execution.ProcessResult;%0A%0A
 import j
@@ -352,16 +352,43 @@
 l.List;%0A
+import java.util.Optional;%0A
 import j
@@ -3647,129 +3647,752 @@
     
-try (LockGuard guard = new LockGuard(lock)) %7B%0A            Runtime.getRuntime().exec(%22pkill -SIGUSR1 -x filedistributor%22);
+String binaryName = %22filedistributor%22;%0A        String command = %22pkill -SIGUSR1 -x %22 + binaryName;%0A        int timeoutSeconds = 10;%0A        try (LockGuard guard = new LockGuard(lock)) %7B%0A            Optional%3CProcessResult%3E result = new ProcessExecutor.Builder(timeoutSeconds).build().execute(command);%0A            if (! result.isPresent()) %7B%0A                throw new RuntimeException(String.format(%22Executing '%25s' timed out%22, command));%0A            %7D else if (result.get().exitCode != 0) %7B%0A                throw new RuntimeException(String.format(%22Error when executing '%25s', exit code %25d: %25s%5Cn%25s%22,%0A                                                         command, result.get().exitCode, result.get().stdOut, result.get().stdErr));%0A            %7D
 %0A   
@@ -4489,28 +4489,22 @@
 ize 
-the filedistributor%22
+%22 + binaryName
 , e)

Handle errors when reinitializing file distributor