This repository was archived by the owner on Feb 26, 2025. It is now read-only.
forked from martint/s3fs
-
Notifications
You must be signed in to change notification settings - Fork 65
Expand file tree
/
Copy pathS3SeekableByteChannel.java
More file actions
157 lines (131 loc) · 5.05 KB
/
Copy pathS3SeekableByteChannel.java
File metadata and controls
157 lines (131 loc) · 5.05 KB
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
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
package com.upplication.s3fs;
import static java.lang.String.format;
import java.io.BufferedInputStream;
import java.io.IOException;
import java.io.InputStream;
import java.nio.ByteBuffer;
import java.nio.channels.SeekableByteChannel;
import java.nio.file.FileAlreadyExistsException;
import java.nio.file.Files;
import java.nio.file.NoSuchFileException;
import java.nio.file.OpenOption;
import java.nio.file.Path;
import java.nio.file.StandardCopyOption;
import java.nio.file.StandardOpenOption;
import java.util.Collections;
import java.util.HashSet;
import java.util.Set;
import org.apache.tika.Tika;
import com.amazonaws.services.s3.Headers;
import com.amazonaws.services.s3.model.ObjectMetadata;
import com.amazonaws.services.s3.model.S3Object;
import com.amazonaws.services.s3.model.SSEAlgorithm;
public class S3SeekableByteChannel implements SeekableByteChannel {
private S3Path path;
private Set<? extends OpenOption> options;
private SeekableByteChannel seekable;
private Path tempFile;
/**
* Open or creates a file, returning a seekable byte channel
*
* @param path the path open or create
* @param options options specifying how the file is opened
* @throws IOException if an I/O error occurs
*/
public S3SeekableByteChannel(S3Path path, Set<? extends OpenOption> options) throws IOException {
this.path = path;
this.options = Collections.unmodifiableSet(new HashSet<>(options));
String key = path.getKey();
boolean exists = path.getFileSystem().provider().exists(path);
if (exists && this.options.contains(StandardOpenOption.CREATE_NEW))
throw new FileAlreadyExistsException(format("target already exists: %s", path));
else if (!exists && !this.options.contains(StandardOpenOption.CREATE_NEW) &&
!this.options.contains(StandardOpenOption.CREATE))
throw new NoSuchFileException(format("target not exists: %s", path));
tempFile = Files.createTempFile("temp-s3-", key.replaceAll("/", "_"));
boolean removeTempFile = true;
try {
if (exists) {
try (S3Object object = path.getFileSystem()
.getClient()
.getObject(path.getFileStore().getBucket().getName(), key)) {
Files.copy(object.getObjectContent(), tempFile, StandardCopyOption.REPLACE_EXISTING);
}
}
Set<? extends OpenOption> seekOptions = new HashSet<>(this.options);
seekOptions.remove(StandardOpenOption.CREATE_NEW);
seekable = Files.newByteChannel(tempFile, seekOptions);
removeTempFile = false;
} finally {
if (removeTempFile) {
Files.deleteIfExists(tempFile);
}
}
}
@Override
public boolean isOpen() {
return seekable.isOpen();
}
@Override
public void close() throws IOException {
try {
if (!seekable.isOpen())
return;
seekable.close();
if (options.contains(StandardOpenOption.DELETE_ON_CLOSE)) {
path.getFileSystem().provider().delete(path);
return;
}
if (options.contains(StandardOpenOption.READ) && options.size() == 1) {
return;
}
sync();
} finally {
Files.deleteIfExists(tempFile);
}
}
/**
* try to sync the temp file with the remote s3 path.
*
* @throws IOException if the tempFile fails to open a newInputStream
*/
protected void sync() throws IOException {
try (InputStream stream = new BufferedInputStream(Files.newInputStream(tempFile))) {
ObjectMetadata metadata = new ObjectMetadata();
metadata.setContentLength(Files.size(tempFile));
if (path.getFileName() != null) {
metadata.setContentType(new Tika().detect(stream, path.getFileName().toString()));
}
if (path.getFileSystem().isServerSideEncryption()) {
metadata.setHeader(Headers.SERVER_SIDE_ENCRYPTION, SSEAlgorithm.getDefault().name());
}
String bucket = path.getFileStore().name();
String key = path.getKey();
path.getFileSystem().getClient().putObject(bucket, key, stream, metadata);
}
}
@Override
public int write(ByteBuffer src) throws IOException {
return seekable.write(src);
}
@Override
public SeekableByteChannel truncate(long size) throws IOException {
return seekable.truncate(size);
}
@Override
public long size() throws IOException {
return seekable.size();
}
@Override
public int read(ByteBuffer dst) throws IOException {
return seekable.read(dst);
}
@Override
public SeekableByteChannel position(long newPosition) throws IOException {
return seekable.position(newPosition);
}
@Override
public long position() throws IOException {
return seekable.position();
}
}