Retrofit+Rxjava下载文件进度的实现

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'

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;  } }; }}

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; }}

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: 取消订阅" ); }}

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); }}

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(); }}

@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); }}

public interface ApiInfo { @Streaming @GET Observable<ResponseBody> download(@Url String url);}

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)); }

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) {  Toast.makeText(App.getInstance(),"下载文件成功",Toast.LENGTH_SHORT).show();  }  @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); }