Support chunked data upload in iOS

Cronet currently accumulates all the upload data from the stream
and sends it on NSStreamEventEndEncountered when using NSMutableURLRequest to post data from a stream.

This CL implements a chunked data uploader to support this
functionality.

Related discuss:
  https://groups.google.com/a/chromium.org/forum/#!topic/net-dev/I02k_dEiq1g

R=kapishnikov@chromium.org

Bug: 755463
Cq-Include-Trybots: master.tryserver.chromium.android:android_cronet_tester;master.tryserver.chromium.mac:ios-simulator-cronet;master.tryserver.chromium.mac:ios-simulator-full-configs
Change-Id: I74aef03f4ffeee8847cd08c870e3d6825a99ccfd
Reviewed-on: https://chromium-review.googlesource.com/813445
Commit-Queue: Andrei Kapishnikov <kapishnikov@chromium.org>
Reviewed-by: Andrei Kapishnikov <kapishnikov@chromium.org>
Reviewed-by: Mark Cogan <marq@chromium.org>
Cr-Commit-Position: refs/heads/master@{#535023}

AUTHORS

@@ -876,6 +876,7 @@ Ye Liu <cbakgly@gmail.com>
 Yeol Park <peary2@gmail.com>
 Yi Shen <yi.shen@samsung.com>
 Yi Sun <ratsunny@gmail.com>
+Yichen Jiang <jiangyichen123@gmail.com>
 Yizhou Jiang <yizhou.jiang@intel.com>
 Yoav Weiss <yoav@yoav.ws>
 Yoav Zilberberg <yoav.zilberberg@gmail.com>

components/cronet/ios/test/cronet_http_test.mm

@@ -25,6 +25,78 @@
 
 #include "url/gurl.h"
 
+namespace {
+
+// The buffer size of the stream for HTTPBodyStream post test.
+const NSUInteger kRequestBodyBufferLength = 1024;
+
+// The buffer size of the stream for HTTPBodyStream post test when
+// testing the stream buffered data size larger than the net stack internal
+// buffer size.
+const NSUInteger kLargeRequestBodyBufferLength = 100 * kRequestBodyBufferLength;
+
+// The body data write times for HTTPBodyStream post test.
+const NSInteger kRequestBodyWriteTimes = 16;
+}
+
+@interface StreamBodyRequestDelegate : NSObject<NSStreamDelegate>
+- (void)setOutputStream:(NSOutputStream*)outputStream;
+- (NSMutableString*)requestBody;
+@end
+@implementation StreamBodyRequestDelegate {
+  NSOutputStream* _stream;
+  NSInteger _count;
+
+  NSMutableString* _requestBody;
+}
+
+- (instancetype)init {
+  _requestBody = [NSMutableString string];
+  return self;
+}
+
+- (void)setOutputStream:(NSOutputStream*)outputStream {
+  _stream = outputStream;
+}
+
+- (NSMutableString*)requestBody {
+  return _requestBody;
+}
+
+- (void)stream:(NSStream*)stream handleEvent:(NSStreamEvent)event {
+  ASSERT_EQ(stream, _stream);
+  switch (event) {
+    case NSStreamEventHasSpaceAvailable: {
+      if (_count < kRequestBodyWriteTimes) {
+        uint8_t buffer[kRequestBodyBufferLength];
+        memset(buffer, 'a' + _count, kRequestBodyBufferLength);
+        NSUInteger bytes_write =
+            [_stream write:buffer maxLength:kRequestBodyBufferLength];
+        ASSERT_EQ(kRequestBodyBufferLength, bytes_write);
+        [_requestBody appendString:[[NSString alloc]
+                                       initWithBytes:buffer
+                                              length:kRequestBodyBufferLength
+                                            encoding:NSUTF8StringEncoding]];
+        ++_count;
+      } else {
+        [_stream close];
+      }
+      break;
+    }
+    case NSStreamEventErrorOccurred:
+    case NSStreamEventEndEncountered: {
+      [_stream close];
+      [_stream setDelegate:nil];
+      [_stream removeFromRunLoop:[NSRunLoop currentRunLoop]
+                         forMode:NSDefaultRunLoopMode];
+      break;
+    }
+    default:
+      break;
+  }
+}
+@end
+
 namespace cronet {
 const char kUserAgent[] = "CronetTest/1.0.0.0";
 
@@ -424,6 +496,115 @@ void TearDown() override {
   ASSERT_TRUE(block_used);
 }
 
+// Verify the chunked request body upload function.
+TEST_F(HttpTest, PostRequestWithBodyStream) {
+  // Create request body stream.
+  CFReadStreamRef read_stream = NULL;
+  CFWriteStreamRef write_stream = NULL;
+  CFStreamCreateBoundPair(NULL, &read_stream, &write_stream,
+                          kRequestBodyBufferLength);
+
+  NSInputStream* input_stream = CFBridgingRelease(read_stream);
+  NSOutputStream* output_stream = CFBridgingRelease(write_stream);
+
+  StreamBodyRequestDelegate* stream_delegate =
+      [[StreamBodyRequestDelegate alloc] init];
+  output_stream.delegate = stream_delegate;
+  [stream_delegate setOutputStream:output_stream];
+
+  dispatch_queue_t queue =
+      dispatch_queue_create("data upload queue", DISPATCH_QUEUE_SERIAL);
+  dispatch_semaphore_t semaphore = dispatch_semaphore_create(0);
+  dispatch_async(queue, ^{
+    [output_stream scheduleInRunLoop:[NSRunLoop currentRunLoop]
+                             forMode:NSDefaultRunLoopMode];
+    [output_stream open];
+
+    [[NSRunLoop currentRunLoop]
+        runUntilDate:[NSDate dateWithTimeIntervalSinceNow:10.0]];
+
+    dispatch_semaphore_signal(semaphore);
+  });
+
+  // Prepare the request.
+  NSURL* url = net::NSURLWithGURL(GURL(TestServer::EchoRequestBodyURL()));
+  NSMutableURLRequest* request = [[NSMutableURLRequest alloc] initWithURL:url];
+  request.HTTPMethod = @"POST";
+  request.HTTPBodyStream = input_stream;
+
+  // Set the request filter to check that the request was handled by the Cronet
+  // stack.
+  __block BOOL block_used = NO;
+  [Cronet setRequestFilterBlock:^(NSURLRequest* req) {
+    block_used = YES;
+    EXPECT_EQ([req URL], url);
+    return YES;
+  }];
+
+  // Send the request and wait for the response.
+  NSURLSessionDataTask* data_task = [session_ dataTaskWithRequest:request];
+  StartDataTaskAndWaitForCompletion(data_task);
+
+  // Verify that the response from the server matches the request body.
+  ASSERT_EQ(nil, [delegate_ error]);
+  NSString* response_body = [delegate_ responseBody];
+  NSMutableString* request_body = [stream_delegate requestBody];
+  ASSERT_STREQ(base::SysNSStringToUTF8(request_body).c_str(),
+               base::SysNSStringToUTF8(response_body).c_str());
+  ASSERT_TRUE(block_used);
+
+  // Wait for the run loop of the child thread exits. Timeout is 5 seconds.
+  dispatch_time_t timeout = dispatch_time(DISPATCH_TIME_NOW, 5 * NSEC_PER_SEC);
+  ASSERT_EQ(0, dispatch_semaphore_wait(semaphore, timeout));
+}
+
+// Verify that the chunked data uploader can correctly handle the request body
+// if the stream contains data length exceed the internal upload buffer.
+TEST_F(HttpTest, PostRequestWithLargeBodyStream) {
+  // Create request body stream.
+  CFReadStreamRef read_stream = NULL;
+  CFWriteStreamRef write_stream = NULL;
+  // 100KB data is written in one time.
+  CFStreamCreateBoundPair(NULL, &read_stream, &write_stream,
+                          kLargeRequestBodyBufferLength);
+
+  NSInputStream* input_stream = CFBridgingRelease(read_stream);
+  NSOutputStream* output_stream = CFBridgingRelease(write_stream);
+  [output_stream open];
+
+  uint8_t buffer[kLargeRequestBodyBufferLength];
+  memset(buffer, 'a', kLargeRequestBodyBufferLength);
+  NSUInteger bytes_write =
+      [output_stream write:buffer maxLength:kLargeRequestBodyBufferLength];
+  ASSERT_EQ(kLargeRequestBodyBufferLength, bytes_write);
+  [output_stream close];
+
+  // Prepare the request.
+  NSURL* url = net::NSURLWithGURL(GURL(TestServer::EchoRequestBodyURL()));
+  NSMutableURLRequest* request = [[NSMutableURLRequest alloc] initWithURL:url];
+  request.HTTPMethod = @"POST";
+  request.HTTPBodyStream = input_stream;
+
+  // Set the request filter to check that the request was handled by the Cronet
+  // stack.
+  __block BOOL block_used = NO;
+  [Cronet setRequestFilterBlock:^(NSURLRequest* req) {
+    block_used = YES;
+    EXPECT_EQ([req URL], url);
+    return YES;
+  }];
+
+  // Send the request and wait for the response.
+  NSURLSessionDataTask* data_task = [session_ dataTaskWithRequest:request];
+  StartDataTaskAndWaitForCompletion(data_task);
+
+  // Verify that the response from the server matches the request body.
+  ASSERT_EQ(nil, [delegate_ error]);
+  NSString* response_body = [delegate_ responseBody];
+  ASSERT_EQ(kLargeRequestBodyBufferLength, [response_body length]);
+  ASSERT_TRUE(block_used);
+}
+
 // iOS Simulator doesn't support changing thread priorities.
 // Therefore, run these tests only on a physical device.
 #if TARGET_OS_SIMULATOR

ios/net/BUILD.gn

@@ -31,6 +31,8 @@ source_set("net") {
   configs += [ "//build/config/compiler:enable_arc" ]
 
   sources = [
+    "chunked_data_stream_uploader.cc",
+    "chunked_data_stream_uploader.h",
     "clients/crn_network_client_protocol.h",
     "cookies/cookie_cache.cc",
     "cookies/cookie_cache.h",
@@ -109,6 +111,7 @@ test("ios_net_unittests") {
   ]
 
   sources = [
+    "chunked_data_stream_uploader_unittest.cc",
     "cookies/cookie_cache_unittest.cc",
     "cookies/cookie_creation_time_manager_unittest.mm",
     "cookies/cookie_store_ios_persistent_unittest.mm",

ios/net/chunked_data_stream_uploader.cc

@@ -0,0 +1,99 @@
+// Copyright (c) 2017 The Chromium Authors. All rights reserved.
+// Use of this source code is governed by a BSD-style license that can be
+// found in the LICENSE file.
+
+#include "ios/net/chunked_data_stream_uploader.h"
+
+#include "base/logging.h"
+#include "net/base/io_buffer.h"
+#include "net/base/net_errors.h"
+
+namespace net {
+
+ChunkedDataStreamUploader::ChunkedDataStreamUploader(Delegate* delegate)
+    : UploadDataStream(true, 0),
+      delegate_(delegate),
+      pending_read_buffer_(nullptr),
+      pending_read_buffer_length_(0),
+      pending_internal_read_(false),
+      is_final_chunk_(false),
+      is_front_of_stream_(true),
+      weak_factory_(this) {
+  DCHECK(delegate_);
+}
+
+ChunkedDataStreamUploader::~ChunkedDataStreamUploader() {}
+
+int ChunkedDataStreamUploader::InitInternal(const NetLogWithSource& net_log) {
+  if (is_front_of_stream_)
+    return OK;
+  else
+    return ERR_FAILED;
+}
+
+int ChunkedDataStreamUploader::ReadInternal(net::IOBuffer* buffer,
+                                            int buffer_length) {
+  DCHECK(buffer);
+  DCHECK_GT(buffer_length, 0);
+  DCHECK(!pending_read_buffer_);
+
+  pending_read_buffer_ = buffer;
+  pending_read_buffer_length_ = buffer_length;
+
+  // Read the stream if input data comes first.
+  return Upload();
+}
+
+void ChunkedDataStreamUploader::ResetInternal() {
+  pending_read_buffer_ = nullptr;
+  pending_read_buffer_length_ = 0;
+  pending_internal_read_ = false;
+  // Internal reset will not affect the external stream data state.
+  is_final_chunk_ = false;
+}
+
+void ChunkedDataStreamUploader::UploadWhenReady(bool is_final_chunk) {
+  is_final_chunk_ = is_final_chunk;
+
+  // Put the data if internal read comes first.
+  if (pending_internal_read_) {
+    Upload();
+  }
+}
+
+int ChunkedDataStreamUploader::Upload() {
+  DCHECK(pending_read_buffer_);
+
+  is_front_of_stream_ = false;
+  int bytes_read = 0;
+
+  if (is_final_chunk_) {
+    SetIsFinalChunk();
+  } else {
+    bytes_read = delegate_->OnRead(pending_read_buffer_->data(),
+                                   pending_read_buffer_length_);
+
+    // NSInputStream can read 0 bytes when hasBytesAvailable is true, so ignore
+    // this piece and let this internal read remain pending.
+    // Still returns ERR_IO_PENDING for other errors because currently it is not
+    // supported to return failure in UploadDataStream::Read(). Handle the
+    // failure in the delegate level.
+    if (bytes_read <= 0) {
+      pending_internal_read_ = true;
+      return ERR_IO_PENDING;
+    }
+  }
+
+  pending_read_buffer_ = nullptr;
+  pending_read_buffer_length_ = 0;
+
+  // When there is a Read() pending, call OnReadCompleted to notify read
+  // completed.
+  if (pending_internal_read_) {
+    pending_internal_read_ = false;
+    OnReadCompleted(bytes_read);
+  }
+  return bytes_read;
+}
+
+}  // namespace net