MultiUploader.java
10.3 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
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
package com.cnlive.demo.rdupload;
import android.os.Handler;
import android.os.Message;
import android.util.Log;
import com.ksyun.ks3.exception.Ks3Error;
import com.ksyun.ks3.model.Part;
import com.ksyun.ks3.model.PartETag;
import com.ksyun.ks3.model.result.CompleteMultipartUploadResult;
import com.ksyun.ks3.model.result.InitiateMultipartUploadResult;
import com.ksyun.ks3.model.result.ListPartsResult;
import com.ksyun.ks3.services.Ks3Client;
import com.ksyun.ks3.services.handler.CompleteMultipartUploadResponseHandler;
import com.ksyun.ks3.services.handler.InitiateMultipartUploadResponceHandler;
import com.ksyun.ks3.services.handler.ListPartsResponseHandler;
import com.ksyun.ks3.services.handler.UploadPartResponceHandler;
import com.ksyun.ks3.services.request.AbortMultipartUploadRequest;
import com.ksyun.ks3.services.request.InitiateMultipartUploadRequest;
import com.ksyun.ks3.services.request.ListPartsRequest;
import com.ksyun.ks3.services.request.UploadPartRequest;
import org.apache.http.Header;
import java.io.File;
import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.CountDownLatch;
/**
* Created by zhaotao on 4/27/17.
*/
public class MultiUploader {
final private String TAG = "MultiUploader";
private String bucketName;
private String key;
private File file;
private long partSize = 5*1024*1024; // 5MB
private String uploadId;
private Ks3Client client;
final static private int INIT_DONE = 0;
final static private int PARTS_DONE = 1;
final static private int COMPLETE_DONE = 2;
final static private int GET_UPLOADED_DONE=3;
final List<PartETag> doneParts = Collections.synchronizedList(new ArrayList());
private void create(Ks3Client client, String bucketName, String key, File file, String uploadId, long partSize){
this.client = client;
this.bucketName = bucketName;
this.key = key;
this.file = file;
this.uploadId = uploadId;
this.partSize = partSize;
}
public MultiUploader(Ks3Client client, String bucketName, String key, File file, String uploadId, long partSize){
create(client, bucketName,key, file, uploadId, partSize);
}
public MultiUploader(Ks3Client client, String bucketName, String key, File file, String uploadId){
create(client, bucketName, key, file, uploadId, partSize);
}
public MultiUploader(Ks3Client client, String bucketName, String key, File file){
create(client, bucketName, key, file, null, partSize);
}
public MultiUploader(Ks3Client client, String bucketName, String key, File file, long partSize){
create(client, bucketName, key, file, null, partSize);
}
public String getKey(){
return key;
}
public String getUploadId(){
return uploadId;
}
abstract class MyUploadPartResponceHandler extends UploadPartResponceHandler{
private String key;
private int partNo;
private String uploadId;
public abstract void onSuccess(int statesCode, Header[] responceHeaders, PartETag result, String key, int partNo, String uploadId);
public abstract void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable throwable
, String key, int partNo, String uploadId);
public abstract void onTaskProgress(double progress, String key, int partNo, String uploadId);
public MyUploadPartResponceHandler(String key, int partNo, String uploadId){
this.key = key;
this.partNo = partNo;
this.uploadId = uploadId;
}
@Override
public void onTaskProgress(double progress) {
onTaskProgress(progress, key, partNo, uploadId);
}
@Override
public void onSuccess(int statesCode, Header[] responceHeaders, PartETag result) {
onSuccess(statesCode, responceHeaders, result, key, partNo, uploadId);
}
@Override
public void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable throwable) {
onFailure(statesCode,error,responceHeaders,response,throwable, key, partNo, uploadId);
}
}
private final MyHandler mHandler = new MyHandler();
class MyHandler extends Handler {
@Override
public void handleMessage(Message msg) {
switch (msg.what) {
case INIT_DONE:
uploadParts();
break;
case PARTS_DONE:
completeUpload();
break;
case GET_UPLOADED_DONE:
List<PartETag> res = (List<PartETag>)msg.obj;
reUpload(res);
break;
default:
break;
}
}
}
/*
return false, if uploadId hasbeen got
*/
public boolean upload(){
if(uploadId != null) return false;
InitiateMultipartUploadRequest request = new InitiateMultipartUploadRequest(bucketName, key);
client.initiateMultipartUpload(request, new InitiateMultipartUploadResponceHandler(){
@Override
public void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable paramThrowable) {
Log.w(TAG, "init multiupload fail, statesCode="+statesCode, paramThrowable);
}
@Override
public void onSuccess(int statesCode, Header[] responceHeaders, InitiateMultipartUploadResult result) {
uploadId = result.getUploadId();
Log.d(TAG, "init multiupload success, uploadId="+uploadId + ",key="+key);
mHandler.sendEmptyMessage(INIT_DONE);
}
});
return true;
}
private List<PartETag> convertPart(List<Part> list){
List<PartETag> res = new ArrayList<PartETag>();
for(Part p : list) {
res.add(new PartETag(p.getPartNumber(), p.getETag()));
}
return res;
}
/*
Get from ks3 server. call list parts API.
*/
public void getUploadedParts(){
final List<PartETag> res = new ArrayList<PartETag>();
final ListPartsResponseHandler listPartsResponseHandler = new ListPartsResponseHandler() {
@Override
public void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable paramThrowable) {
Log.w(TAG, "list parts, statesCode="+statesCode, paramThrowable);
}
@Override
public void onSuccess(int statesCode, Header[] responceHeaders, ListPartsResult listPartsResult) {
res.addAll(convertPart(listPartsResult.getParts()));
if(! listPartsResult.isTruncated())
mHandler.sendMessage(Message.obtain(mHandler, GET_UPLOADED_DONE, res));
else
Log.e(TAG, "File size too largs. You may not use phone to upload");
}
};
client.listParts(bucketName, key, uploadId, listPartsResponseHandler);
}
public List<Integer> getLeftParts(List<PartETag> uploadedParts){
List<Integer> res = new ArrayList<Integer>();
long start = 0L;
int partNumber = 1;
Set<Integer> set = new HashSet<Integer>();
for(PartETag p : uploadedParts){
set.add(p.getPartNumber());
}
while(start<file.length()){
if(! set.contains(partNumber))
res.add(partNumber);
partNumber++;
start += partSize;
}
return res;
}
public void reUpload(){
if(uploadId == null) {
Log.i(TAG, "no upload id, cannot reupload");
return;
}
getUploadedParts();
}
public void reUpload(final List<PartETag> uploadedParts){
final List<Integer> leftParts = getLeftParts(uploadedParts);
final int N = leftParts.size() + uploadedParts.size();
doneParts.addAll(uploadedParts);
if(leftParts.isEmpty())
mHandler.sendEmptyMessage(PARTS_DONE);
for(Integer partNumber : leftParts){
long offset = partNumber * partSize - partSize;
UploadPartRequest uploadPartRequest = new UploadPartRequest(bucketName, key,uploadId, file,
offset, partNumber, Math.min(file.length()-offset, partSize));
client.uploadPart(uploadPartRequest, new MyUploadPartResponceHandler(key, partNumber, uploadId){
@Override
public void onSuccess(int statesCode, Header[] responceHeaders, PartETag result, String key, int partNo, String uploadId) {
result.setPartNumber(partNo);
doneParts.add(result);
if(doneParts.size() == N)
mHandler.sendEmptyMessage(PARTS_DONE);
}
@Override
public void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable throwable,
String key, int partNo, String uploadId) {
// you may record this failure info in file, database, or backend
Log.w(TAG, "upload part fail, uploadId="+uploadId+",key="+key+",partNo="+partNo, throwable);
}
@Override
public void onTaskProgress(double progress, String key, int partNo, String uploadId) {
if(progress>=99)
Log.i(TAG, "progress:"+progress+",key="+key+",partNo"+partNo);
}
});
}
}
private void uploadParts(){
reUpload(new ArrayList<PartETag>());
}
private void completeUpload(){
client.completeMultipartUpload(bucketName,key,uploadId,doneParts,new CompleteMultipartUploadResponseHandler(){
@Override
public void onFailure(int statesCode, Ks3Error error, Header[] responceHeaders, String response, Throwable paramThrowable) {
Log.w(TAG, "complete upload fail, statusCode="+statesCode, paramThrowable);
}
@Override
public void onSuccess(int statesCode, Header[] responceHeaders, CompleteMultipartUploadResult result) {
Log.i(TAG, "complete upload, key="+key);
}
});
}
}