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
127
128
129
130
|
package org.openslx.bwlp.sat.fileserv;
import java.io.File;
import java.io.FileNotFoundException;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.Iterator;
import java.util.List;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import org.apache.log4j.Logger;
import org.openslx.bwlp.sat.util.Constants;
import org.openslx.bwlp.sat.util.Formatter;
import org.openslx.bwlp.thrift.iface.ImageDetailsRead;
import org.openslx.bwlp.thrift.iface.TTransferRejectedException;
import org.openslx.bwlp.thrift.iface.UserInfo;
import org.openslx.filetransfer.Downloader;
import org.openslx.filetransfer.IncomingEvent;
import org.openslx.filetransfer.Listener;
import org.openslx.filetransfer.Uploader;
public class FileServer implements IncomingEvent {
private static final Logger LOGGER = Logger.getLogger(FileServer.class);
/**
* Listener for incoming unencrypted connections
*/
private final Listener plainListener = new Listener(this, null, 9092, Constants.TRANSFER_TIMEOUT); // TODO: Config
private final ThreadPoolExecutor transferPool = new ThreadPoolExecutor(2, Constants.MAX_UPLOADS + Constants.MAX_DOWNLOADS, 1, TimeUnit.MINUTES, new ArrayBlockingQueue<Runnable>(1));
/**
* All currently running uploads, indexed by token
*/
private final Map<String, ActiveUpload> uploads = new ConcurrentHashMap<>();
private static final FileServer globalInstance = new FileServer();
private FileServer() {
}
public static FileServer instance() {
return globalInstance;
}
public boolean start() {
boolean ret = plainListener.start();
// TODO: Start SSL listener too
return ret;
}
@Override
public void incomingDownloadRequest(Uploader uploader) throws IOException {
// TODO Auto-generated method stub
}
@Override
public void incomingUploadRequest(Downloader downloader) throws IOException {
String token = downloader.getToken();
LOGGER.info("Incoming filetransfer with token " + token);
ActiveUpload upload = uploads.get(token);
if (upload == null) {
LOGGER.warn("Unknown token " + token);
downloader.cancel();
return;
}
upload.addConnection(downloader, transferPool);
}
/**
* Get an upload instance by given token.
*
* @param uploadToken
* @return
*/
public ActiveUpload getUploadByToken(String uploadToken) {
return uploads.get(uploadToken);
}
public String createNewUserUpload(UserInfo owner, ImageDetailsRead image, long fileSize, List<ByteBuffer> sha1Sums)
throws TTransferRejectedException {
Iterator<ActiveUpload> it = uploads.values().iterator();
int activeUploads = 0;
while (it.hasNext()) {
ActiveUpload upload = it.next();
if (upload.isComplete()) {
// TODO: Check age (short timeout) and remove
continue;
} else {
// Check age (long timeout) and remove
}
activeUploads++;
}
if (activeUploads > Constants.MAX_UPLOADS)
throw new TTransferRejectedException("Server busy. Too many running uploads.");
File destinationFile = null;
do {
destinationFile = Formatter.getTempImageName();
} while (destinationFile.exists());
destinationFile.getParentFile().mkdirs();
String key = UUID.randomUUID().toString();
ActiveUpload upload;
try {
upload = new ActiveUpload(key, owner, image, destinationFile, fileSize, sha1Sums);
} catch (FileNotFoundException e) {
LOGGER.error("Could not open destination file for writing", e);
throw new TTransferRejectedException("Destination file not writable!");
}
uploads.put(key, upload);
return key;
}
public int getPlainPort() {
return plainListener.getPort();
}
public int getSslPort() {
return 0; // TODO
}
}
|
