前言
最近在學習Retrofit，雖然Retrofit沒有提供文件下載進度的回調，但是Retrofit底層依賴的是OkHttp，實際上所需要的實現OkHttp對下載進度的監聽，在OkHttp的官方Demo中，有一個Progress.java的文件，顧名思義。點我查看。
準備工作
本文采用Dagger2，Retrofit，RxJava。
|
1
2
3
4
5
6
7
8
9
10
|
compile'com.squareup.retrofit2:retrofit:2.0.2'compile'com.squareup.retrofit2:converter-gson:2.0.2'compile'com.squareup.retrofit2:adapter-rxjava:2.0.2'//dagger2compile'com.google.dagger:dagger:2.6'apt'com.google.dagger:dagger-compiler:2.6'//RxJavacompile'io.reactivex:rxandroid:1.2.0'compile'io.reactivex:rxjava:1.1.5'compile'com.jakewharton.rxbinding:rxbinding:0.4.0' |
改造ResponseBody
okHttp3默認的ResponseBody因為不知道進度的相關信息，所以需要對其進行改造。可以使用接口監聽進度信息。這里采用的是RxBus發送FileLoadEvent對象實現對下載進度的實時更新。這里先講改造的ProgressResponseBody。
|
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
|
public class ProgressResponseBody extends ResponseBody { private ResponseBody responseBody; private BufferedSource bufferedSource; public ProgressResponseBody(ResponseBody responseBody) { this.responseBody = responseBody; } @Override public MediaType contentType() { return responseBody.contentType(); } @Override public long contentLength() { return responseBody.contentLength(); } @Override public BufferedSource source() { if (bufferedSource == null) { bufferedSource = Okio.buffer(source(responseBody.source())); } return bufferedSource; } private Source source(Source source) { return new ForwardingSource(source) { long bytesReaded = 0; @Override public long read(Buffer sink, long byteCount) throws IOException { long bytesRead = super.read(sink, byteCount); bytesReaded += bytesRead == -1 ? 0 : bytesRead; //實時發送當前已讀取的字節和總字節 RxBus.getInstance().post(new FileLoadEvent(contentLength(), bytesReaded)); return bytesRead; } }; }} |
呃，OKIO相關知識我也正在學，這個是從官方Demo中copy的代碼，只不過中間使用了RxBus實時發送FileLoadEvent對象。
FileLoadEvent
FileLoadEvent很簡單，包含了當前已加載進度和文件總大小。
|
1
2
3
4
5
6
7
8
9
10
11
12
13
14
|
public class FileLoadEvent { long total; long bytesLoaded; public long getBytesLoaded() { return bytesLoaded; } public long getTotal() { return total; } public FileLoadEvent(long total, long bytesLoaded) { this.total = total; this.bytesLoaded = bytesLoaded; }} |
RxBus
RxBus 名字看起來像一個庫，但它并不是一個庫，而是一種模式，它的思想是使用 RxJava 來實現了 EventBus ，而讓你不再需要使用OTTO或者 EventBus。點我查看詳情。
|
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
|
public class RxBus { private static volatile RxBus mInstance; private SerializedSubject<Object, Object> mSubject; private HashMap<String, CompositeSubscription> mSubscriptionMap; /** * PublishSubject只會把在訂閱發生的時間點之后來自原始Observable的數據發射給觀察者 * Subject同時充當了Observer和Observable的角色，Subject是非線程安全的，要避免該問題， * 需要將 Subject轉換為一個 SerializedSubject ，上述RxBus類中把線程非安全的PublishSubject包裝成線程安全的Subject。 */ private RxBus() { mSubject = new SerializedSubject<>(PublishSubject.create()); } /** * 單例 雙重鎖 * @return */ public static RxBus getInstance() { if (mInstance == null) { synchronized (RxBus.class) { if (mInstance == null) { mInstance = new RxBus(); } } } return mInstance; } /** * 發送一個新的事件 * @param o */ public void post(Object o) { mSubject.onNext(o); } /** * 根據傳遞的 eventType 類型返回特定類型(eventType)的 被觀察者 * @param type * @param <T> * @return */ public <T> Observable<T> tObservable(final Class<T> type) { //ofType操作符只發射指定類型的數據，其內部就是filter+cast return mSubject.ofType(type); } public <T> Subscription doSubscribe(Class<T> type, Action1<T> next, Action1<Throwable> error) { return tObservable(type) .onBackpressureBuffer() .subscribeOn(Schedulers.io()) .observeOn(AndroidSchedulers.mainThread()) .subscribe(next, error); } public void addSubscription(Object o, Subscription subscription) { if (mSubscriptionMap == null) { mSubscriptionMap = new HashMap<>(); } String key = o.getClass().getName(); if (mSubscriptionMap.get(key) != null) { mSubscriptionMap.get(key).add(subscription); } else { CompositeSubscription compositeSubscription = new CompositeSubscription(); compositeSubscription.add(subscription); mSubscriptionMap.put(key, compositeSubscription); // Log.e("air", "addSubscription:訂閱成功 " ); } } public void unSubscribe(Object o) { if (mSubscriptionMap == null) { return; } String key = o.getClass().getName(); if (!mSubscriptionMap.containsKey(key)) { return; } if (mSubscriptionMap.get(key) != null) { mSubscriptionMap.get(key).unsubscribe(); } mSubscriptionMap.remove(key); //Log.e("air", "unSubscribe: 取消訂閱" ); }} |
FileCallBack
那么，重點來了。代碼其實有5個方法需要重寫，好吧，其實這些方法可以精簡一下。其中progress()方法有兩個參數，progress和total,分別表示文件已下載的大小和總大小，我們將這兩個參數不斷更新到UI上就行了。
|
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
|
public abstract class FileCallBack<T> { private String destFileDir; private String destFileName; public FileCallBack(String destFileDir, String destFileName) { this.destFileDir = destFileDir; this.destFileName = destFileName; subscribeLoadProgress(); } public abstract void onSuccess(T t); public abstract void progress(long progress, long total); public abstract void onStart(); public abstract void onCompleted(); public abstract void onError(Throwable e); public void saveFile(ResponseBody body) { InputStream is = null; byte[] buf = new byte[2048]; int len; FileOutputStream fos = null; try { is = body.byteStream(); File dir = new File(destFileDir); if (!dir.exists()) { dir.mkdirs(); } File file = new File(dir, destFileName); fos = new FileOutputStream(file); while ((len = is.read(buf)) != -1) { fos.write(buf, 0, len); } fos.flush(); unsubscribe(); //onCompleted(); } catch (FileNotFoundException e) { e.printStackTrace(); } catch (IOException e) { e.printStackTrace(); } finally { try { if (is != null) is.close(); if (fos != null) fos.close(); } catch (IOException e) { Log.e("saveFile", e.getMessage()); } } } /** * 訂閱加載的進度條 */ public void subscribeLoadProgress() { Subscription subscription = RxBus.getInstance().doSubscribe(FileLoadEvent.class, new Action1<FileLoadEvent>() { @Override public void call(FileLoadEvent fileLoadEvent) { progress(fileLoadEvent.getBytesLoaded(),fileLoadEvent.getTotal()); } }, new Action1<Throwable>() { @Override public void call(Throwable throwable) { //TODO 對異常的處理 } }); RxBus.getInstance().addSubscription(this, subscription); } /** * 取消訂閱，防止內存泄漏 */ public void unsubscribe() { RxBus.getInstance().unSubscribe(this); }} |
開始下載
使用自己的ProgressResponseBody
通過OkHttpClient的攔截器去攔截Response，并將我們的ProgressReponseBody設置進去監聽進度。
|
1
2
3
4
5
6
7
8
9
|
public class ProgressInterceptor implements Interceptor { @Override public Response intercept(Chain chain) throws IOException { Response originalResponse = chain.proceed(chain.request()); return originalResponse.newBuilder() .body(new ProgressResponseBody(originalResponse.body())) .build(); }} |
構建Retrofit
|
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
|
@Modulepublic class ApiModule { @Provides @Singleton public OkHttpClient provideClient() { OkHttpClient client = new OkHttpClient.Builder() .addInterceptor(new ProgressInterceptor()) .build(); return client; } @Provides @Singleton public Retrofit provideRetrofit(OkHttpClient client){ Retrofit retrofit = new Retrofit.Builder() .client(client) .baseUrl(Constant.HOST) .addCallAdapterFactory(RxJavaCallAdapterFactory.create()) .addConverterFactory(GsonConverterFactory.create()) .build(); return retrofit; } @Provides @Singleton public ApiInfo provideApiInfo(Retrofit retrofit){ return retrofit.create(ApiInfo.class); } @Provides @Singleton public ApiManager provideApiManager(Application application, ApiInfo apiInfo){ return new ApiManager(application,apiInfo); }} |
請求接口
|
1
2
3
4
5
|
public interface ApiInfo { @Streaming @GET Observable<ResponseBody> download(@Url String url);} |
執行請求
|
1
2
3
4
5
6
7
8
9
10
11
12
13
|
public void load(String url, final FileCallBack<ResponseBody> callBack){ apiInfo.download(url) .subscribeOn(Schedulers.io())//請求網絡 在調度者的io線程 .observeOn(Schedulers.io()) //指定線程保存文件 .doOnNext(new Action1<ResponseBody>() { @Override public void call(ResponseBody body) { callBack.saveFile(body); } }) .observeOn(AndroidSchedulers.mainThread()) //在主線程中更新ui .subscribe(new FileSubscriber<ResponseBody>(application,callBack)); } |
在presenter層中執行網絡請求。
通過V層依賴注入的presenter對象調用請求網絡，請求網絡后調用V層更新UI的操作。
|
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
|
public void load(String url){ String fileName = "app.apk"; String fileStoreDir = Environment.getExternalStorageDirectory().getAbsolutePath(); Log.e(TAG, "load: "+fileStoreDir.toString() ); FileCallBack<ResponseBody> callBack = new FileCallBack<ResponseBody>(fileStoreDir,fileName) { @Override public void onSuccess(final ResponseBody responseBody) { } @Override public void progress(long progress, long total) { iHomeView.update(total,progress); } @Override public void onStart() { iHomeView.showLoading(); } @Override public void onCompleted() { iHomeView.hideLoading(); } @Override public void onError(Throwable e) { //TODO: 對異常的一些處理 e.printStackTrace(); } }; apiManager.load(url, callBack); } |
踩到的坑。
依賴的Retrofit版本一定要保持一致！！！說多了都是淚啊。
保存文件時要使用RxJava的doOnNext操作符，后續更新UI的操作切換到UI線程。
總結
看似代碼很多，其實過程并不復雜：
在保存文件時，調用ForwardingSource的read方法，通過RxBus發送實時的FileLoadEvent對象。
FileCallBack訂閱RxBus發送的FileLoadEvent。通過接收到FileLoadEvent中的下載進度和文件總大小對UI進行更新。
在下載保存文件完成后，取消訂閱，防止內存泄漏。
Demo地址：https://github.com/AirMiya/DownloadDemo
原文鏈接：http://www.jianshu.com/p/060d55fc1c82
