Fix multipart streams need to be recreated when doing retry

Refactor the CBLMultipartUploader's init method to accept a block for creating a new mutipart writer object instead of a multipart writer object that cannot be reused when doing retry.

#665

Source/CBLMultipartUploader.h

@@ -9,16 +9,22 @@
 #import "CBLRemoteRequest.h"
 #import "CBLMultipartWriter.h"
 
+/** The signature of the mulipart writer block called by the CBLMultipartUploader to 
+    get a CBLMultipartWriter object. When the CBLMultipartUploader is doing retry, it
+    will call the block to get a new writer object. It cannot reuse the old writer object
+    as all of the streams have already been opened and cannot be reopened. */
+typedef CBLMultipartWriter* (^CBLMultipartUploaderMultipartWriterBlock)(void);
 
 @interface CBLMultipartUploader : CBLRemoteRequest
 {
     @private
-    CBLMultipartWriter* _multipartWriter;
+    CBLMultipartUploaderMultipartWriterBlock _writer;
+    CBLMultipartWriter* _currentWriter;
 }
 
 - (instancetype) initWithURL: (NSURL *)url
-                    streamer: (CBLMultipartWriter*)streamer
               requestHeaders: (NSDictionary *) requestHeaders
+             multipartWriter: (CBLMultipartUploaderMultipartWriterBlock)writer
                 onCompletion: (CBLRemoteRequestCompletionBlock)onCompletion;
 
 @end

Source/CBLMultipartUploader.m

@@ -15,56 +15,58 @@
 
 #import "CBLMultipartUploader.h"
 
+@interface CBLMultipartUploader ()
+
+@end
 
 @implementation CBLMultipartUploader
 
 - (instancetype) initWithURL: (NSURL *)url
-                    streamer: (CBLMultipartWriter*)writer
               requestHeaders: (NSDictionary *) requestHeaders
-                onCompletion: (CBLRemoteRequestCompletionBlock)onCompletion
-{
+             multipartWriter: (CBLMultipartUploaderMultipartWriterBlock)writer
+                onCompletion: (CBLRemoteRequestCompletionBlock)onCompletion {
     Assert(writer);
     self = [super initWithMethod: @"PUT" 
                              URL: url 
-                            body: writer
+                            body: nil
                   requestHeaders: requestHeaders 
                     onCompletion: onCompletion];
     if (self) {
-        _multipartWriter = writer;
-        // It's important to set a Content-Length header -- without this, CFNetwork won't know the
-        // length of the body stream, so it has to send the body chunked. But unfortunately CouchDB
-        // doesn't correctly parse chunked multipart bodies:
-        // https://issues.apache.org/jira/browse/COUCHDB-1403
-        SInt64 length = _multipartWriter.length;
-        Assert(length >= 0, @"HTTP multipart upload body has indeterminate length");
-        [_request setValue: $sprintf(@"%lld", length) forHTTPHeaderField: @"Content-Length"];
+        _writer = [writer copy];
     }
     return self;
 }
 
 
+- (void) start {
+    _currentWriter = _writer();
 
+    // It's important to set a Content-Length header -- without this, CFNetwork won't know the
+    // length of the body stream, so it has to send the body chunked. But unfortunately CouchDB
+    // doesn't correctly parse chunked multipart bodies:
+    // https://issues.apache.org/jira/browse/COUCHDB-1403
+    SInt64 length = _currentWriter.length;
+    Assert(length >= 0, @"HTTP multipart upload body has indeterminate length");
+    [_request setValue: $sprintf(@"%lld", length) forHTTPHeaderField: @"Content-Length"];
 
-- (void) start {
-    [_multipartWriter openForURLRequest: _request];
+    [_currentWriter openForURLRequest: _request];
     [super start];
 }
 
 
 - (NSInputStream *)connection:(NSURLConnection *)connection
-            needNewBodyStream:(NSURLRequest *)request
-{
+            needNewBodyStream:(NSURLRequest *)request {
     LogTo(CBLRemoteRequest, @"%@: Needs new body stream, resetting writer...", self);
-    [_multipartWriter close];
-    return [_multipartWriter openForInputStream];
+    [_currentWriter close];
+    return [_currentWriter openForInputStream];
 }
 
 
 - (void)connection:(NSURLConnection *)connection didFailWithError:(NSError *)error {
     if ($equal(error.domain, NSURLErrorDomain) && error.code == NSURLErrorRequestBodyStreamExhausted) {
         // The connection is complaining that the body input stream closed prematurely.
         // Check whether this is because the multipart writer got an error on _its_ input stream:
-        NSError* writerError = _multipartWriter.error;
+        NSError* writerError = _currentWriter.error;
         if (writerError)
             error = writerError;
     }

Source/CBL_Pusher.m

@@ -414,8 +414,7 @@ CBLStatus CBLStatusFromBulkDocsResponseItem(NSDictionary* item) {
         return kCBLStatusUpstreamError;
 }
 
-
-- (BOOL) uploadMultipartRevision: (CBL_Revision*)rev {
+- (CBLMultipartWriter*)multipartWriterForRevision: (CBL_Revision*)rev {
     // Find all the attachments with "follows" instead of a body, and put 'em in a multipart stream.
     // It's important to scan the _attachments entries in the same order in which they will appear
     // in the JSON, because CouchDB expects the MIME bodies to appear in that same order (see #133).
@@ -435,7 +434,7 @@ - (BOOL) uploadMultipartRevision: (CBL_Revision*)rev {
                 NSData* json = [CBJSONEncoder canonicalEncoding: rev.properties error: &error];
                 if (error) {
                     Warn(@"%@: Creating canonical JSON data got an error: %@", self, error);
-                    return NO;
+                    return nil;
                 }
 
                 if (self.canSendCompressedRequests)
@@ -456,10 +455,17 @@ - (BOOL) uploadMultipartRevision: (CBL_Revision*)rev {
                                                              named: attachmentName
                                                             status: &status];
             if (!attachmentObj)
-                return NO;
+                return nil;
             [bodyStream addStream: attachmentObj.contentStream length: attachmentObj->length];
         }
     }
+
+    return bodyStream;
+}
+
+- (BOOL) uploadMultipartRevision: (CBL_Revision*)rev {
+    // Pre-creating the body stream and check if it's available or not.
+    __block CBLMultipartWriter* bodyStream = [self multipartWriterForRevision: rev];
     if (!bodyStream)
         return NO;
 
@@ -470,8 +476,16 @@ - (BOOL) uploadMultipartRevision: (CBL_Revision*)rev {
     NSString* path = $sprintf(@"%@?new_edits=false", CBLEscapeURLParam(rev.docID));
     __block CBLMultipartUploader* uploader = [[CBLMultipartUploader alloc]
                                   initWithURL: CBLAppendToURL(_remote, path)
-                                     streamer: bodyStream
                                requestHeaders: self.requestHeaders
+                              multipartWriter:^CBLMultipartWriter *{
+                                  CBLMultipartWriter* writer = bodyStream;
+                                  // Reset to nil so the writer will get regenerated if the block
+                                  // gets re-called (e.g. when retrying).
+                                  bodyStream = nil;
+                                  if (!writer)
+                                      writer = [self multipartWriterForRevision: rev];
+                                  return writer;
+                              }
                                  onCompletion: ^(CBLMultipartUploader* result, NSError *error) {
                   [self removeRemoteRequest: uploader];
                   if (error) {