基于Retrofit+Rxjava实现带进度显示的下载文件

//rxjava
 compile 'io.reactivex:rxjava:latest.release'
 compile 'io.reactivex:rxandroid:latest.release'
 //network - squareup
 compile 'com.squareup.retrofit2:retrofit:latest.release'
 compile 'com.squareup.retrofit2:adapter-rxjava:latest.release'
 compile 'com.squareup.okhttp3:okhttp:latest.release'
 compile 'com.squareup.okhttp3:logging-interceptor:latest.release'

public interface downloadprogresslistener {
 void update(long bytesread, long contentlength, boolean done);
}

public class downloadprogressresponsebody extends responsebody {

 private responsebody responsebody;
 private downloadprogresslistener progresslistener;
 private bufferedsource bufferedsource;

 public downloadprogressresponsebody(responsebody responsebody,
          downloadprogresslistener progresslistener) {
  this.responsebody = responsebody;
  this.progresslistener = progresslistener;
 }

 @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 totalbytesread = 0l;

   @override
   public long read(buffer sink, long bytecount) throws ioexception {
    long bytesread = super.read(sink, bytecount);
    // read() returns the number of bytes read, or -1 if this source is exhausted.
    totalbytesread += bytesread != -1 ? bytesread : 0;

    if (null != progresslistener) {
     progresslistener.update(totalbytesread, responsebody.contentlength(), bytesread == -1);
    }
    return bytesread;
   }
  };

 }
}

public class downloadprogressinterceptor implements interceptor {

 private downloadprogresslistener listener;

 public downloadprogressinterceptor(downloadprogresslistener listener) {
  this.listener = listener;
 }

 @override
 public response intercept(chain chain) throws ioexception {
  response originalresponse = chain.proceed(chain.request());

  return originalresponse.newbuilder()
    .body(new downloadprogressresponsebody(originalresponse.body(), listener))
    .build();
 }
}

public class download implements parcelable {

 private int progress;
 private long currentfilesize;
 private long totalfilesize;

 public int getprogress() {
  return progress;
 }

 public void setprogress(int progress) {
  this.progress = progress;
 }

 public long getcurrentfilesize() {
  return currentfilesize;
 }

 public void setcurrentfilesize(long currentfilesize) {
  this.currentfilesize = currentfilesize;
 }

 public long gettotalfilesize() {
  return totalfilesize;
 }

 public void settotalfilesize(long totalfilesize) {
  this.totalfilesize = totalfilesize;
 }

 @override
 public int describecontents() {
  return 0;
 }

 @override
 public void writetoparcel(parcel dest, int flags) {
  dest.writeint(this.progress);
  dest.writelong(this.currentfilesize);
  dest.writelong(this.totalfilesize);
 }

 public download() {
 }

 protected download(parcel in) {
  this.progress = in.readint();
  this.currentfilesize = in.readlong();
  this.totalfilesize = in.readlong();
 }

 public static final parcelable.creator<download> creator = new parcelable.creator<download>() {
  @override
  public download createfromparcel(parcel source) {
   return new download(source);
  }

  @override
  public download[] newarray(int size) {
   return new download[size];
  }
 };
}

public interface downloadservice {

 @streaming
 @get
 observable<responsebody> download(@url string url);
}

public class downloadapi {
 private static final string tag = "downloadapi";
 private static final int default_timeout = 15;
 public retrofit retrofit;


 public downloadapi(string url, downloadprogresslistener listener) {

  downloadprogressinterceptor interceptor = new downloadprogressinterceptor(listener);

  okhttpclient client = new okhttpclient.builder()
    .addinterceptor(interceptor)
    .retryonconnectionfailure(true)
    .connecttimeout(default_timeout, timeunit.seconds)
    .build();


  retrofit = new retrofit.builder()
    .baseurl(url)
    .client(client)
    .addcalladapterfactory(rxjavacalladapterfactory.create())
    .build();
 }

 public void downloadapk(@nonnull string url, final file file, subscriber subscriber) {
  log.d(tag, "downloadapk: " + url);

  retrofit.create(downloadservice.class)
    .download(url)
    .subscribeon(schedulers.io())
    .unsubscribeon(schedulers.io())
    .map(new func1<responsebody, inputstream>() {
     @override
     public inputstream call(responsebody responsebody) {
      return responsebody.bytestream();
     }
    })
    .observeon(schedulers.computation())
    .doonnext(new action1<inputstream>() {
     @override
     public void call(inputstream inputstream) {
      try {
       fileutils.writefile(inputstream, file);
      } catch (ioexception e) {
       e.printstacktrace();
       throw new customizeexception(e.getmessage(), e);
      }
     }
    })
    .observeon(androidschedulers.mainthread())
    .subscribe(subscriber);
 }


}

public class downloadservice extends intentservice {
 private static final string tag = "downloadservice";

 private notificationcompat.builder notificationbuilder;
 private notificationmanager notificationmanager;


 private string apkurl = "http://download.fir.im/v2/app/install/595c5959959d6901ca0004ac?download_token=1a9dfa8f248b6e45ea46bc5ed96a0a9e&source=update";

 public downloadservice() {
  super("downloadservice");
 }

 @override
 protected void onhandleintent(intent intent) {
  notificationmanager = (notificationmanager) getsystemservice(context.notification_service);

  notificationbuilder = new notificationcompat.builder(this)
    .setsmallicon(r.mipmap.ic_download)
    .setcontenttitle("download")
    .setcontenttext("downloading file")
    .setautocancel(true);

  notificationmanager.notify(0, notificationbuilder.build());

  download();
 }

 private void download() {
  downloadprogresslistener listener = new downloadprogresslistener() {
   @override
   public void update(long bytesread, long contentlength, boolean done) {
    download download = new download();
    download.settotalfilesize(contentlength);
    download.setcurrentfilesize(bytesread);
    int progress = (int) ((bytesread * 100) / contentlength);
    download.setprogress(progress);

    sendnotification(download);
   }
  };
  file outputfile = new file(environment.getexternalstoragepublicdirectory
    (environment.directory_downloads), "file.apk");
  string baseurl = stringutils.gethostname(apkurl);

  new downloadapi(baseurl, listener).downloadapk(apkurl, outputfile, new subscriber() {
   @override
   public void oncompleted() {
    downloadcompleted();
   }

   @override
   public void onerror(throwable e) {
    e.printstacktrace();
    downloadcompleted();
    log.e(tag, "onerror: " + e.getmessage());
   }

   @override
   public void onnext(object o) {

   }
  });
 }

 private void downloadcompleted() {
  download download = new download();
  download.setprogress(100);
  sendintent(download);

  notificationmanager.cancel(0);
  notificationbuilder.setprogress(0, 0, false);
  notificationbuilder.setcontenttext("file downloaded");
  notificationmanager.notify(0, notificationbuilder.build());
 }

 private void sendnotification(download download) {

  sendintent(download);
  notificationbuilder.setprogress(100, download.getprogress(), false);
  notificationbuilder.setcontenttext(
    stringutils.getdatasize(download.getcurrentfilesize()) + "/" +
      stringutils.getdatasize(download.gettotalfilesize()));
  notificationmanager.notify(0, notificationbuilder.build());
 }

 private void sendintent(download download) {

  intent intent = new intent(mainactivity.message_progress);
  intent.putextra("download", download);
  localbroadcastmanager.getinstance(downloadservice.this).sendbroadcast(intent);
 }

 @override
 public void ontaskremoved(intent rootintent) {
  notificationmanager.cancel(0);
 }
}

public class mainactivity extends appcompatactivity {

 public static final string message_progress = "message_progress";

 private appcompatbutton btn_download;
 private progressbar progress;
 private textview progress_text;


 private broadcastreceiver broadcastreceiver = new broadcastreceiver() {
  @override
  public void onreceive(context context, intent intent) {

   if (intent.getaction().equals(message_progress)) {

    download download = intent.getparcelableextra("download");
    progress.setprogress(download.getprogress());
    if (download.getprogress() == 100) {

     progress_text.settext("file download complete");

    } else {

     progress_text.settext(stringutils.getdatasize(download.getcurrentfilesize())
       +"/"+
       stringutils.getdatasize(download.gettotalfilesize()));

    }
   }
  }
 };

 @override
 protected void oncreate(bundle savedinstancestate) {
  super.oncreate(savedinstancestate);
  setcontentview(r.layout.activity_main);
  btn_download = (appcompatbutton) findviewbyid(r.id.btn_download);
  progress = (progressbar) findviewbyid(r.id.progress);
  progress_text = (textview) findviewbyid(r.id.progress_text);

  registerreceiver();

  btn_download.setonclicklistener(new view.onclicklistener() {
   @override
   public void onclick(view view) {
    intent intent = new intent(mainactivity.this, downloadservice.class);
    startservice(intent);
   }
  });
 }

 private void registerreceiver() {

  localbroadcastmanager bmanager = localbroadcastmanager.getinstance(this);
  intentfilter intentfilter = new intentfilter();
  intentfilter.addaction(message_progress);
  bmanager.registerreceiver(broadcastreceiver, intentfilter);

 }
}