RxJ2+Retrofit+OkHttp 学习分享（3）

文件上传
这次我打算用作者的代码来解释，因为之前已经分析了。整个流程下面，可以飙车了。直接添加一个文件操作，，应该怎么做。
上代码
一如既往，我们从 MainActivity 中开始处理问题：

 private void uploadeDo() {
//      不比多说这个就是一个文件
        File file = new File("/storage/emulated/0/Download/11.jpg");
//      对文件主体直接添加文件标识符
        RequestBody requestBody = RequestBody.create(MediaType.parse("image/jpeg"), file);
//      这里new ProgressRequestBody 文件，从名字就感觉是个body体。一个很新的的监听UploadProgressListener
        MultipartBody.Part part = MultipartBody.Part.createFormData("file_name", file.getName(), new ProgressRequestBody(requestBody,
                new UploadProgressListener() {
                    @Override
                    public void onProgress(long currentBytesCount, long totalBytesCount) {
                        tvMsg.setText("提示:上传中");
                        progressBar.setMax((int) totalBytesCount);
                        progressBar.setProgress((int) currentBytesCount);
                    }
                }));
//     和之前讲过的一样的一个，监听模式 辩护的是，这个，UploadApi 是新的。
        UploadApi uplaodApi = new UploadApi(httpOnNextListener, this);
        uplaodApi.setPart(part);
        HttpManager manager = HttpManager.getInstance();
        manager.doHttpDeal(uplaodApi);
    }


    /**
     * 上传回调
     */
    HttpOnNextListener httpOnNextListener = new HttpOnNextListener<UploadResulte>() {
        @Override
        public void onNext(UploadResulte o) {
            tvMsg.setText("成功");
            Glide.with(MainActivity.this).load(o.getHeadImgUrl()).skipMemoryCache(true).into(img);
        }

        @Override
        public void onError(Throwable e) {
            super.onError(e);
            tvMsg.setText("失败：" + e.toString());
        }

    };

接下来，先看首次出现 ProgressRequestBody 类

// 继承的是OKhttp的RequestBody 懂http的应该知道，body就是主体内容
public class ProgressRequestBody extends RequestBody {
    //实际的待包装请求体
    private final RequestBody requestBody;
    /*
    * 进度回调接口 一个新的接口，但是看名字就知道，进度条监听
    如果没有忘记上面的，直接new 对这个这个实例化的一个监听吗？对了。  就是他，一个简单的函数回调
    * **/
    private final UploadProgressListener progressListener;
    //包装完成的BufferedSink  这里是一个io流的缓冲区
    private BufferedSink bufferedSink;

public ProgressRequestBody(RequestBody requestBody, UploadProgressListener progressListener) {
        this.requestBody = requestBody;
        this.progressListener = progressListener;
    }
    /**
     * 重写调用实际的响应体的contentType
     * @return MediaType
     */
    @Override
    public MediaType contentType() {
        return requestBody.contentType();
    }
    /**
     * 重写调用实际的响应体的contentLength
     * @return contentLength
     * @throws IOException 异常
     */
    @Override
    public long contentLength() throws IOException {
        return requestBody.contentLength();
    }
    /**
     * 重写进行写入
     * @param sink BufferedSink
     * @throws IOException 异常
     */
    @Override
    public void writeTo(BufferedSink sink) throws IOException {
        if (null == bufferedSink) {
// sink调用的是下面的方法 
            bufferedSink = Okio.buffer(sink(sink));
        }
        requestBody.writeTo(bufferedSink);
        //必须调用flush，否则最后一部分数据可能不会被写入
        bufferedSink.flush();
    }
    /**
     * 写入，回调进度接口
     * @param sink Sink 看源码他是两个接口的实现类，其实他还是个缓冲区的意思，你会好奇，
                    BufferedSink 和 sink什么关心， BufferedSink 继承 Sink
                    你理解他是流操作，绝对没有问题。 
                    如果你知道，pipe的化那就更好了。理解他是通道 刷新一下，就提交
     * @return Sink
     */
    private Sink sink(Sink sink) {
        return new ForwardingSink(sink) {
            //当前写入字节数
            long writtenBytesCount = 0L;
            //总字节长度，避免多次调用contentLength()方法
            long totalBytesCount = 0L;
            @Override
            public void write(Buffer source, long byteCount) throws IOException {
                super.write(source, byteCount);
                //增加当前写入的字节数
                writtenBytesCount += byteCount;
                //获得contentLength的值，后续不再调用
                if (totalBytesCount == 0) {
                    totalBytesCount = contentLength();
                }
//这里比较留，重新会到主线程，数显进度条。回调就在这里。            Observable.just(writtenBytesCount).observeOn(AndroidSchedulers.mainThread()).subscribe(new Action1<Long>() {
                    @Override
                    public void call(Long aLong) {
                        progressListener.onProgress(writtenBytesCount, totalBytesCount);
                    }
                });
            }
        };
    }
}

以上的body已经准备完毕，面对是UploadApi 这个心累

//文件依旧是集成baseApi 这个不多解释
public class UploadApi extends BaseApi {
    /*需要上传的文件*/
    private MultipartBody.Part part;
    public UploadApi(HttpOnNextListener listener, RxAppCompatActivity rxAppCompatActivity) {
        super(listener, rxAppCompatActivity);
        setShowProgress(true);
        setMothed("AppFiftyToneGraph/videoLink");
    }

    public MultipartBody.Part getPart() {
        return part;
    }

    public void setPart(MultipartBody.Part part) {
        this.part = part;
    }

    @Override
    public Observable getObservable(Retrofit retrofit) {
//一个上传的接口。
        HttpUploadService service = retrofit.create(HttpUploadService.class);
//接口的两个参数。并没有什么特别的，OK全部搞定。
        RequestBody uid= RequestBody.create(MediaType.parse("text/plain"), "4811420");
        RequestBody key = RequestBody.create(MediaType.parse("text/plain"), "cfed6cc8caad0d79ea56d917376dc4df");
        return service.uploadImage(uid,key,getPart());
    }

}

以上是文件所有上传代码，内容并不多·~~ 自己测试的时候，需要更换文件地址。
库原作者的，githup下载地址 返现库原作者的服务器，不能上传。这个，，我也很郁闷。。。。