1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
|
// Copyright 2017 Yahoo Holdings. Licensed under the terms of the Apache 2.0 license. See LICENSE in the project root.
package com.yahoo.vespa.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"));
}
//---------------- 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.DEBUG, "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 exception: " + 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.queueForAsyncDownload(fileReferences);
req.returnValues().add(new Int32Value(0));
}
}
|
