[Android] Replaced ForwardingSink with OutputStream to fix progress

Using a ForwardingSink, an IllegalStateException was thrown in Okio's
RealBufferedSink when attempting to write to a sink that was closed.
Additionally, it did not send updates for non-input stream request bodies.
Replacing with an OutputStream-based sink prevents the crash by throwing an
IOException instead, and fixes the progress updates. Also, now get the content
length before writing to avoid incorrect total size for input streams.
Addresses issues: 10423/11016.

ReactAndroid/src/main/java/com/facebook/react/modules/network/ProgressRequestBody.java

@@ -9,60 +9,75 @@
 
 package com.facebook.react.modules.network;
 
+import com.facebook.common.internal.CountingOutputStream;
+
 import java.io.IOException;
+
 import okhttp3.MediaType;
 import okhttp3.RequestBody;
 import okio.BufferedSink;
-import okio.Buffer;
-import okio.Sink;
-import okio.ForwardingSink;
 import okio.Okio;
+import okio.Sink;
 
 public class ProgressRequestBody extends RequestBody {
 
   private final RequestBody mRequestBody;
   private final ProgressListener mProgressListener;
   private BufferedSink mBufferedSink;
+  private long mContentLength = 0L;
 
   public ProgressRequestBody(RequestBody requestBody, ProgressListener progressListener) {
-      mRequestBody = requestBody;
-      mProgressListener = progressListener;
+    mRequestBody = requestBody;
+    mProgressListener = progressListener;
   }
 
   @Override
   public MediaType contentType() {
-      return mRequestBody.contentType();
+    return mRequestBody.contentType();
   }
 
   @Override
   public long contentLength() throws IOException {
-      return mRequestBody.contentLength();
+    if (mContentLength == 0) {
+      mContentLength = mRequestBody.contentLength();
+    }
+    return mContentLength;
   }
 
   @Override
   public void writeTo(BufferedSink sink) throws IOException {
-      if (mBufferedSink == null) {
-          mBufferedSink = Okio.buffer(sink(sink));
-      }
-      mRequestBody.writeTo(mBufferedSink);
-      mBufferedSink.flush();
+    if (mBufferedSink == null) {
+      mBufferedSink = Okio.buffer(outputStreamSink(sink));
+    }
+
+    // contentLength changes for input streams, since we're using inputStream.available(),
+    // so get the length before writing to the sink
+    contentLength();
+
+    mRequestBody.writeTo(mBufferedSink);
+    mBufferedSink.flush();
   }
 
-  private Sink sink(Sink sink) {
-      return new ForwardingSink(sink) {
-          long bytesWritten = 0L;
-          long contentLength = 0L;
+  private Sink outputStreamSink(BufferedSink sink) {
+    return Okio.sink(new CountingOutputStream(sink.outputStream()) {
+      @Override
+      public void write(byte[] data, int offset, int byteCount) throws IOException {
+        super.write(data, offset, byteCount);
+        sendProgressUpdate();
+      }
 
-          @Override
-          public void write(Buffer source, long byteCount) throws IOException {
-              super.write(source, byteCount);
-              if (contentLength == 0) {
-                  contentLength = contentLength();
-              }
-              bytesWritten += byteCount;
-              mProgressListener.onProgress(
-                bytesWritten, contentLength, bytesWritten == contentLength);
-          }
-      };
+      @Override
+      public void write(int data) throws IOException {
+        super.write(data);
+        sendProgressUpdate();
+      }
+
+      private void sendProgressUpdate() throws IOException {
+        long bytesWritten = getCount();
+        long contentLength = contentLength();
+        mProgressListener.onProgress(
+          bytesWritten, contentLength, bytesWritten == contentLength);
+      }
+    });
   }
 }