@@ -0,0 +1,2802 @@
+package com.jagsaund.rxuploader.rx;%0A%0Aimport android.support.annotation.NonNull;%0Aimport android.support.annotation.VisibleForTesting;%0Aimport com.jagsaund.rxuploader.job.Status;%0Aimport java.io.File;%0Aimport java.io.FileInputStream;%0Aimport java.io.FileNotFoundException;%0Aimport java.io.IOException;%0Aimport java.io.InputStream;%0Aimport okhttp3.MediaType;%0Aimport okhttp3.RequestBody;%0Aimport okio.BufferedSink;%0Aimport okio.Okio;%0Aimport okio.Source;%0Aimport rx.Emitter;%0A%0A/**%0A * Transforms a %7B@linkplain RequestBody%7D into one that is reactive and will emit progress updates%0A * to the provide %7B@code Emitter%7D. Progress updates are emitted as %7B@linkplain Status%7D items.%0A */%0Apublic class RxRequestBody extends RequestBody %7B%0A    public static final long BUFFER_SIZE = 8 * 1024;%0A%0A    @NonNull private final String jobId;%0A    @NonNull private final MediaType mediaType;%0A    @NonNull private final InputStream inputStream;%0A    @NonNull private final Emitter%3CStatus%3E progressEmitter;%0A%0A    private final long length;%0A%0A    @NonNull%0A    public static RequestBody create(@NonNull Emitter%3CStatus%3E progressEmitter,%0A            @NonNull String jobId, @NonNull File file, @NonNull String mimeType)%0A            throws FileNotFoundException %7B%0A        final MediaType mediaType = MediaType.parse(mimeType);%0A        final InputStream inputStream = new FileInputStream(file);%0A        final long length = file.length();%0A%0A        return new RxRequestBody(progressEmitter, jobId, mediaType, inputStream, length);%0A    %7D%0A%0A    @VisibleForTesting%0A    RxRequestBody(@NonNull Emitter%3CStatus%3E progressEmitter, @NonNull String jobId,%0A            @NonNull MediaType mediaType, @NonNull InputStream inputStream, long length) %7B%0A        this.progressEmitter = progressEmitter;%0A        this.jobId = jobId;%0A        this.mediaType = mediaType;%0A        this.inputStream = inputStream;%0A        this.length = length;%0A    %7D%0A%0A    @NonNull%0A    @Override%0A    public MediaType contentType() %7B%0A        return mediaType;%0A    %7D%0A%0A    @Override%0A    public long contentLength() %7B%0A        return length;%0A    %7D%0A%0A    @Override%0A    public void writeTo(@NonNull BufferedSink sink) throws IOException %7B%0A        try (final Source source = Okio.source(inputStream)) %7B%0A            final long total = contentLength();%0A%0A            long remaining = total;%0A            long size = Math.min(BUFFER_SIZE, remaining);%0A%0A            progressEmitter.onNext(Status.createSending(jobId, 0));%0A%0A            while (remaining %3E 0) %7B%0A                sink.write(source, size);%0A%0A                remaining = Math.max(0, remaining - size);%0A                size = Math.min(BUFFER_SIZE, remaining);%0A%0A                final int progress = (int) (100 - ((float) remaining / total) * 100);%0A                progressEmitter.onNext(Status.createSending(jobId, progress));%0A            %7D%0A        %7D%0A    %7D%0A%7D%0A

Add Reactive Request Body