首先展示效果如下：
本demo使用的是rxjava2, rxjava1也是可以使用的；主要核心代码如下：
1.创建个listener
/**
* Created by JokAr on 2017/3/6.
*/
public interface UploadListener {
void onRequestProgress(
long bytesWritten,
long contentLength);
}
2.获取上传进度
/**
* Created by JokAr on 2017/3/6.
*/
public class CountingRequestBody extends RequestBody {
private RequestBody mRequestBody;
private UploadListener mUploadListener;
private CountingSink mCountingSink;
public CountingRequestBody(RequestBody requestBody, UploadListener uploadListener) {
mRequestBody = requestBody;
mUploadListener = uploadListener;
}
@Override
public MediaType
contentType() {
return mRequestBody.contentType();
}
@Override
public long contentLength()
throws IOException {
try {
return mRequestBody.contentLength();
}
catch (IOException e) {
e.printStackTrace();
return -
1;
}
}
@Override
public void writeTo(BufferedSink sink)
throws IOException {
BufferedSink bufferedSink;
mCountingSink =
new CountingSink(sink);
bufferedSink = Okio.buffer(mCountingSink);
mRequestBody.writeTo(bufferedSink);
bufferedSink.flush();
}
class CountingSink extends ForwardingSink {
private long bytesWritten =
0;
public CountingSink(Sink delegate) {
super(delegate);
}
@Override
public void write(Buffer source,
long byteCount)
throws IOException {
super.write(source, byteCount);
bytesWritten += byteCount;
mUploadListener.onRequestProgress(bytesWritten, contentLength());
}
}
}
3.添加拦截器
/**
* Created by JokAr on 2017/3/6.
*/
public class UpLoadProgressInterceptor implements Interceptor {
private UploadListener mUploadListener;
public UpLoadProgressInterceptor(UploadListener uploadListener) {
mUploadListener = uploadListener;
}
@Override
public Response
intercept(Chain chain)
throws IOException {
Request request = chain.request();
if(
null == request.body()){
return chain.proceed(request);
}
Request build = request.newBuilder()
.method(request.method(),
new CountingRequestBody(request.body(),
mUploadListener))
.build();
return chain.proceed(build);
}
}
4.使用方法
public static Retrofit getRetrofit(UploadListener listener) {
UpLoadProgressInterceptor interceptor = new UpLoadProgressInterceptor(listener)
OkHttpClient client = new OkHttpClient
.Builder()
.addInterceptor(interceptor)
.retryOnConnectionFailure(true)
.connectTimeout(DEFAULT_TIMEOUT, TimeUnit
.SECONDS)
.build()
return new Retrofit
.Builder()
.baseUrl(BASE_URL +
"/")
.client(client)
.addConverterFactory(GsonConverterFactory
.create())
.addCallAdapterFactory(RxJava2CallAdapterFactory
.create())
.build()
}
单个文件上传：
@Override
public void uploadFile(File file,
UploadListener listener,
@NonNull LifecycleTransformer transformer,
@NonNull
final UploadFileCallBack callBack) {
if (file !=
null && file.exists()) {
RequestBody requestFile =
RequestBody.create(MediaType.parse(
"application/otcet-stream"), file);
MultipartBody.Part body =
MultipartBody.Part.createFormData(
"file", file.getName(), requestFile);
UploadNetWorkConfig.getRetrofit(listener)
.create(UploadFileService.class)
.uploadHead(body)
.compose(transformer)
.subscribeOn(Schedulers.io())
.unsubscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(
new HttpResultFunc())
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
new ResourceObserver<UploadEntities>() {
@Override
protected void onStart() {
super.onStart();
callBack.onStart();
}
@Override
public void onNext(UploadEntities data) {
callBack.result(data);
}
@Override
public void onError(Throwable e) {
callBack.requestError(e);
}
@Override
public void onComplete() {
callBack.complete();
}
});
}
}
多个文件上传：
/**
* 上传多个文件
*
* @param pathList
* @param listener
* @param transformer
* @param callBack
*/
@Override
public void uploadMultiFile(String[] pathList,
UploadListener listener,
@NonNull LifecycleTransformer transformer,
@NonNull
final UploadMultiFileCallBack callBack) {
if (pathList !=
null && pathList.length >
0) {
ArrayList<Observable> observables =
new ArrayList<>();
for (String path : pathList) {
File file =
new File(path);
if (file.exists()) {
RequestBody requestFile =
RequestBody.create(MediaType.parse(
"application/otcet-stream"), file);
MultipartBody.Part body =
MultipartBody.Part.createFormData(
"file", file.getName(), requestFile);
Observable<String> observable = UploadNetWorkConfig.getRetrofit(listener)
.create(UploadFileService.class)
.uploadHead(body)
.compose(transformer)
.subscribeOn(Schedulers.io())
.unsubscribeOn(Schedulers.io())
.observeOn(Schedulers.newThread())
.map(
new HttpResultFunc());
observables.add(observable);
}
file =
null;
}
Observable[] observables1 =
new Observable[observables.size()];
observables1 = (Observable[]) observables.toArray();
observables =
null;
Observable.zipArray(
new Function<Object[], ArrayList<UploadEntities>>() {
/**
* Apply some calculation to the input value and return some other value.
*
* @param objects the input value
* @return the output value
* @throws Exception on error
*/
@Override
public ArrayList<UploadEntities>
apply(@NonNull Object[] objects)
throws Exception {
ArrayList<UploadEntities> arrayList =
new ArrayList<>();
for (Object object : objects) {
arrayList.add((UploadEntities) object);
}
return arrayList;
}
},
true,
1, observables1)
.observeOn(AndroidSchedulers.mainThread())
.subscribe(
new ResourceObserver<ArrayList<UploadEntities>>() {
@Override
protected void onStart() {
super.onStart();
callBack.onStart();
}
@Override
public void onNext(ArrayList<UploadEntities> data) {
callBack.result(data);
}
@Override
public void onError(Throwable e) {
callBack.requestError(e);
}
@Override
public void onComplete() {
callBack.complete();
}
});
}
}
主要代码就是这些了；完整demo请看这里： https://github.com/MichaelJokAr/RxUpload 如果你觉得对你有用，欢迎start
