OkHttp upload progress is not in sync with the actual upload

Question

I'm trying to track the progress of an upload using OkHttp. I have created a custom RequestBody with the following body (courtesy of this answer) which writes to sink and publishes the progress.

public class CountingFileRequestBody extends RequestBody {
    private static final String TAG = "CountingFileRequestBody";

    private final ProgressListener listener;
    private final String key;
    private final MultipartBody multipartBody;
    protected CountingSink mCountingSink;

    public CountingFileRequestBody(MultipartBody multipartBody,
                                   String key,
                                   ProgressListener listener) {
        this.multipartBody = multipartBody;
        this.listener = listener;
        this.key = key;
    }

    @Override
    public long contentLength() throws IOException {
        return multipartBody.contentLength();
    }

    @Override
    public MediaType contentType() {
        return multipartBody.contentType();
    }

    @Override
    public void writeTo(BufferedSink sink) throws IOException {
        mCountingSink = new CountingSink(sink);
        BufferedSink bufferedSink = Okio.buffer(mCountingSink);
        multipartBody.writeTo(bufferedSink);
        bufferedSink.flush();
    }

    public interface ProgressListener {
        void transferred(String key, int num);
    }

    protected final class CountingSink extends ForwardingSink {
        private long bytesWritten = 0;

        public CountingSink(Sink delegate) {
            super(delegate);
        }

        @Override
        public void write(Buffer source, long byteCount) throws IOException {
            bytesWritten += byteCount;
            listener.transferred(key, (int) (100F * bytesWritten / contentLength()));
            super.write(source, byteCount);
            delegate().flush(); // I have added this line to manually flush the sink
        }
    }

}

The problem here is, the data is written to the sink immediately without actually sending the buffered bytes to the server. Meaning my progress reaches the end long before the actual upload.
Note: Some say that the sink needs to be flushed on every iteration for bytes to be actually uploaded but it's not working for me.

Answer 1

I know this is an old post, but for someone with same problem, I had adapted a helper class (ProgressOutputStream) from this lib: https://github.com/lizhangqu/CoreProgress and my working code is as follow: (working for uploading files and for uploading json too)

import java.io.IOException;

import okhttp3.MediaType;
import okhttp3.RequestBody;
import okio.BufferedSink;
import okio.Okio;

public class UploadProgressRequestBody extends RequestBody {
    private final RequestBody requestBody;
    private final ProgressListener progressListener;

    public UploadProgressRequestBody(RequestBody requestBody) {
        this.requestBody = requestBody;
        this.progressListener = getDefaultProgressListener();
    }

    @Override public MediaType contentType() {
        return requestBody.contentType();
    }

    @Override public long contentLength() {
        try {
            return requestBody.contentLength();
        } catch (IOException e) {
            e.printStackTrace();
        }
        return -1;
    }

    @Override
    public void writeTo(BufferedSink sink) throws IOException {
        if (progressListener == null) {
            requestBody.writeTo(sink);
            return;
        }
        ProgressOutputStream progressOutputStream = new ProgressOutputStream(sink.outputStream(), progressListener, contentLength());
        BufferedSink progressSink = Okio.buffer(Okio.sink(progressOutputStream));
        requestBody.writeTo(progressSink);
        progressSink.flush();
    }

    interface ProgressListener {
        void update(long bytesWritten, long contentLength);
    }

    private ProgressListener getDefaultProgressListener(){
        ProgressListener progressListener = new UploadProgressRequestBody.ProgressListener() {
            @Override public void update(long bytesRead, long contentLength) {
                System.out.println("bytesRead: "+bytesRead);
                System.out.println("contentLength: "+contentLength);
                System.out.format("%d%% done\n", (100 * bytesRead) / contentLength);
            }
        };

        return progressListener;
    }

}

===========

import java.io.IOException;
import java.io.OutputStream;

class ProgressOutputStream extends OutputStream {
    private final OutputStream stream;
    private final UploadProgressRequestBody.ProgressListener listener;

    private long total;
    private long totalWritten;

    ProgressOutputStream(OutputStream stream, UploadProgressRequestBody.ProgressListener listener, long total) {
        this.stream = stream;
        this.listener = listener;
        this.total = total;
    }

    @Override
    public void write(byte[] b, int off, int len) throws IOException {
        this.stream.write(b, off, len);
        if (this.total < 0) {
            this.listener.update(-1, -1);
            return;
        }
        if (len < b.length) {
            this.totalWritten += len;
        } else {
            this.totalWritten += b.length;
        }
        this.listener.update(this.totalWritten, this.total);
    }

    @Override
    public void write(int b) throws IOException {
        this.stream.write(b);
        if (this.total < 0) {
            this.listener.update(-1, -1);
            return;
        }
        this.totalWritten++;
        this.listener.update(this.totalWritten, this.total);
    }

    @Override
    public void close() throws IOException {
        if (this.stream != null) {
            this.stream.close();
        }
    }

    @Override
    public void flush() throws IOException {
        if (this.stream != null) {
            this.stream.flush();
        }
    }
}

==========

    OkHttpClient client = new OkHttpClient.Builder()
    .addNetworkInterceptor(new Interceptor() {
        @Override public Response intercept(Chain chain) throws IOException {
            Request originalRequest = chain.request();

            if (originalRequest.body() == null) {
                return chain.proceed(originalRequest);
            }

            Request progressRequest = originalRequest.newBuilder()
                    .method(originalRequest.method(),
                            new UploadProgressRequestBody(originalRequest.body()))
                    .build();

            return chain.proceed(progressRequest);
        }
    }).build();

Retrofit retrofit = new Retrofit.Builder()
        .baseUrl(baseUrl)
        .client(client)
        .addConverterFactory(GsonConverterFactory.create())
        .build();