Retrofit Rxjava实现下载文件 基于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);

 }
}