...
| Code Block | ||
|---|---|---|
| ||
@Override
public void processInOpen() throws IOException {
if (!preload) {
final File file = new File(filename);
try {
in = new FileInputStream(file);
fireOnConnect();
} catch (Exception e) {
fireOnDisconnect();
throw e;
}
} else {
fis = new FileInputStream(filename);
FileChannel channel = fis.getChannel();
long size = channel.size();
double x = size / (double) Integer.MAX_VALUE;
int n = (int) Math.ceil(x);
ByteBuffer buffers[] = new ByteBuffer[n];
for (int i = 0; i < n; i++) {
buffers[i] = ByteBuffer.allocateDirect(Integer.MAX_VALUE);
channel.read(buffers[i]);
buffers[i].rewind();
}
in = createInputStream(buffers);
fireOnConnect();
}
}
@Override
public void processInClose() throws IOException {
super.processInClose();
if (fis != null) {
fis.close();
}
}
@Override
public void processOutOpen() throws IOException {
final File file = new File(filename);
try {
out = new FileOutputStream(file, append);
fireOnConnect();
} catch (Exception e) {
fireOnDisconnect();
throw e;
}
}
@Override
public void processOutClose() throws IOException {
fireOnDisconnect();
out.flush();
out.close();
} |
Source-Generic PUll
...
Generic Pull
After the connection is inialized, the framework tries to retrieve data from the TransportHandler. To be generic, we decided to use an InputStream for sources and an OutputStream for sinks. So the following methods need to be overwritten (Remark: It its not necessary to implement both methods, if the TransportHandler e.g. should only be used for sources):
| Code Block | ||
|---|---|---|
| ||
public InputStream getInputStream();
public OutputStream getOutputStream(); |
A typical implementation in FileHandler:
| Code Block | ||
|---|---|---|
| ||
@Override
public InputStream getInputStream() {
return in;
}
@Override
public OutputStream getOutputStream() {
return out;
} |
Source-Generic Push
Sink-Generic Pull
Sink-Generic Push
...