feat(storage): Restart & delete resumable upload

generated/google-apis-storage_v1/lib/google/apis/storage_v1/service.rb

@@ -2858,6 +2858,8 @@ def get_object_iam_policy(bucket, object, generation: nil, user_project: nil, fi
         #   IO stream or filename containing content to upload
         # @param [String] content_type
         #   Content type of the uploaded content.
+        # @param @param [IO, String] upload_id
+        #   Unique upload Id for ongoing resumable upload       
         # @param [Google::Apis::RequestOptions] options
         #   Request-specific options
         #
@@ -2870,14 +2872,15 @@ def get_object_iam_policy(bucket, object, generation: nil, user_project: nil, fi
         # @raise [Google::Apis::ServerError] An error occurred on the server and the request can be retried
         # @raise [Google::Apis::ClientError] The request is invalid and should not be retried without modification
         # @raise [Google::Apis::AuthorizationError] Authorization is required
-        def insert_object(bucket, object_object = nil, content_encoding: nil, if_generation_match: nil, if_generation_not_match: nil, if_metageneration_match: nil, if_metageneration_not_match: nil, kms_key_name: nil, name: nil, predefined_acl: nil, projection: nil, user_project: nil, fields: nil, quota_user: nil, user_ip: nil, upload_source: nil, content_type: nil, options: nil, &block)
+        def insert_object(bucket, object_object = nil, content_encoding: nil, if_generation_match: nil, if_generation_not_match: nil, if_metageneration_match: nil, if_metageneration_not_match: nil, kms_key_name: nil, name: nil, predefined_acl: nil, projection: nil, user_project: nil, fields: nil, quota_user: nil, user_ip: nil, upload_source: nil, content_type: nil, upload_id: nil, options: nil, &block)
 
           if upload_source.nil?
             command = make_simple_command(:post, 'b/{bucket}/o', options)
           else
             command = make_storage_upload_command(:post, 'b/{bucket}/o', options)
             command.upload_source = upload_source
             command.upload_content_type = content_type
+            command.upload_id = upload_id
           end
           command.request_representation = Google::Apis::StorageV1::Object::Representation
           command.request_object = object_object
@@ -3848,7 +3851,17 @@ def delete_project_hmac_key(project_id, access_id, user_project: nil, fields: ni
           command.query['userIp'] = user_ip unless user_ip.nil?
           execute_or_queue_command(command, &block)
         end
-
+
+        # Deletes Resumable upload
+        def delete_ongoing_resumable_upload(bucket, upload_source, upload_id, options: nil)
+          command = make_storage_upload_command(:post, 'b/{bucket}/o', options)
+          command.upload_source = upload_source
+          command.upload_id = upload_id
+          command.params['bucket'] = bucket unless bucket.nil?
+          command.delete_upload = true
+          execute_or_queue_command(command)
+        end
+
         # Retrieves an HMAC key's metadata
         # @param [String] project_id
         #   Project ID owning the service account of the requested key.

google-apis-core/lib/google/apis/core/storage_upload.rb

@@ -18,7 +18,6 @@
 require 'stringio'
 require 'tempfile'
 require 'mini_mime'
-
 module Google
   module Apis
     module Core
@@ -37,6 +36,14 @@ class StorageUploadCommand < ApiCommand
         # @return [String, File, #read]
         attr_accessor :upload_source
 
+        # Unique upload_id of a resumable upload
+        # @return [String]
+        attr_accessor :upload_id
+
+        # Boolean Value to specify is a resumable upload is to be deleted or not
+        # @return [Boolean]
+        attr_accessor :delete_upload
+
         # Content type of the upload material
         # @return [String]
         attr_accessor :upload_content_type
@@ -96,9 +103,18 @@ def execute(client)
           prepare!
           opencensus_begin_span
           @upload_chunk_size = options.upload_chunk_size
+          if upload_id.nil?
+            do_retry :initiate_resumable_upload, client
+          elsif delete_upload && !upload_id.nil?
+            make_resumabple_upload_url
+            cancel_resumable_upload(client)
+          else
+            make_resumabple_upload_url
+            reinitiate_resumable_upload(client)
+          end
 
-          do_retry :initiate_resumable_upload, client
           while @upload_incomplete
+
             res = do_retry :send_upload_command, client
           end
           res
@@ -125,12 +141,30 @@ def initiate_resumable_upload(client)
                          body: body,
                          header: request_header,
                          follow_redirect: true)
+
           result = process_response(response.status_code, response.header, response.body)
           success(result)
         rescue => e
           error(e, rethrow: true)
         end
 
+        # Reinitiating resumable upload
+
+        def reinitiate_resumable_upload(client)
+          logger.debug { sprintf('Restarting resumable upload command to %s', url) }
+          check_resumable_upload_status client
+          upload_io.pos = @offset
+          rescue => e
+            error(e, rethrow: true)
+        end
+
+        def make_resumabple_upload_url
+          query_params = query.dup
+          query_params['uploadType'] = RESUMABLE
+          query_params['upload_id'] = upload_id
+          resumable_upload_params = query_params.map { |key, value| "#{key}=#{value}" }.join('&')
+          @upload_url = "#{url}&#{resumable_upload_params}"
+        end
         # Send the actual content
         #
         # @param [HTTPClient] client
@@ -152,9 +186,7 @@ def send_upload_command(client)
             else
               StringIO.new(upload_io.read(current_chunk_size))
             end
-
           response = client.put(@upload_url, body: chunk_body, header: request_header, follow_redirect: true)
-
           result = process_response(response.status_code, response.header, response.body)
           @upload_incomplete = false if response.status_code.eql? OK_STATUS
           @offset += current_chunk_size if @upload_incomplete
@@ -182,6 +214,51 @@ def process_response(status, header, body)
           super(status, header, body)
         end
 
+        def check_resumable_upload_status(client)
+          # Setting up request header
+          request_header = header.dup
+          request_header[CONTENT_RANGE_HEADER] = "bytes */#{upload_io.size}"
+          request_header[CONTENT_LENGTH_HEADER] = '0'
+          # Initiating call
+          response = client.put(@upload_url, header: request_header, follow_redirect: true)
+          case response.code.to_i
+          when 308
+            if response.headers['Range']
+              range = response.headers['Range']
+              @offset = range ? range.split('-').last.to_i + 1 : 0
+              puts "Upload is incomplete. Bytes uploaded: #{response.headers['Range']}"
+            else
+              puts 'No bytes uploaded yet.'
+            end
+            @upload_incomplete = true
+          when 499
+            # Upload in canceled
+            @upload_incomplete = false
+          when 200, 201
+            # Upload is complete.
+            @upload_incomplete = false
+          else
+            puts "Unexpected response: #{response.code} - #{response.body}"
+            @upload_incomplete = true
+          end
+        end
+
+        # Cancel resumable upload
+        def cancel_resumable_upload(client)
+          # Setting up request header
+          request_header = header.dup
+          request_header[CONTENT_LENGTH_HEADER] = '0'
+          # Initiating call
+          response = client.delete(@upload_url, header: request_header, follow_redirect: true)
+          case response.code.to_i
+          when 499
+            @close_io_on_finish = true
+            @upload_incomplete = false
+          else
+            puts "Failed to cancel upload session. Response: #{response.code} - #{response.body}"
+          end
+        end
+
         def streamable?(upload_source)
           upload_source.is_a?(IO) || upload_source.is_a?(StringIO) || upload_source.is_a?(Tempfile)
         end

google-apis-core/lib/google/apis/options.rb

@@ -41,7 +41,8 @@ module Apis
       :quota_project,
       :query,
       :add_invocation_id_header,
-      :upload_chunk_size)
+      :upload_chunk_size
+    )
 
     # General client options
     class ClientOptions

google-apis-core/spec/google/apis/core/storage_upload_spec.rb

@@ -119,7 +119,7 @@
       stub_request(:put, 'https://www.googleapis.com/zoo/animals')
         .with(headers: { 'Content-Range' => 'bytes 11-21/22' })
         .to_return(body: %(OK))
-    end    
+    end
 
     it 'should make requests multiple times' do
       command.options.upload_chunk_size = 11
@@ -129,6 +129,111 @@
     end
   end
 
+  context('restart resumable upload with upload_url') do
+    let(:file) { StringIO.new('Hello world' * 3) }
+    let(:upload_id) { 'TestId' }
+    let(:upload_url) { "https://www.googleapis.com/zoo/animals?uploadType=resumable&upload_id=#{upload_id}" }
+
+    before(:example) do
+      stub_request(:put, upload_url)
+        .with(
+          headers: {
+            'Content-Length' => '0',
+            'Content-Range' => 'bytes */33'
+          }
+        )
+        .to_return(
+          status: [308, 'Resume Incomplete'],
+          headers: { 'Range' => 'bytes=0-21' }
+        )
+    end
+
+    before(:example) do
+      stub_request(:put, upload_url)
+        .with(headers: { 'Content-Range' => 'bytes 22-32/33' })
+        .to_return(body: %(OK))
+    end
+
+    it 'should restart a resumable upload' do
+      command.options.upload_chunk_size = 11
+      command.upload_id = upload_id
+      command.execute(client)
+      expect(a_request(:put, upload_url)
+        .with(body: 'Hello world')).to have_been_made
+    end
+  end
+
+  context('should not restart resumable upload if upload is completed') do
+    let(:file) { StringIO.new('Hello world' * 3) }
+    let(:upload_id) {"TestId"}
+    let(:upload_url) { "https://www.googleapis.com/zoo/animals?uploadType=resumable&upload_id=#{upload_id}" }
+
+    before(:example) do
+      stub_request(:put, upload_url)
+        .with(
+          headers: {
+            'Content-Length' => '0',
+            'Content-Range' => 'bytes */33'
+          }
+        )
+        .to_return(status: 200, headers: { 'Range' => 'bytes=0-32' })
+    end
+
+    before(:example) do
+      stub_request(:put, upload_url)
+        .with(headers: { 'Content-Range' => 'bytes */33' })
+        .to_return(status: 200)
+    end
+
+    it 'should not restart a upload' do
+      command.options.upload_chunk_size = 11
+      command.upload_id = upload_id
+      command.execute(client)
+      expect(a_request(:put, upload_url)
+        .with(body: 'Hello world')).to have_not_been_made
+    end
+  end
+
+  context('delete resumable upload with upload_id') do
+    let(:file) { StringIO.new('Hello world' * 3) }
+    let(:upload_id) { 'TestId' }
+    let(:upload_url) { "https://www.googleapis.com/zoo/animals?uploadType=resumable&upload_id=#{upload_id}" }
+
+
+    before(:example) do
+      stub_request(:delete, upload_url)
+        .with(headers: { 'Content-Length' => '0' })
+        .to_return(status: [499])
+    end
+
+    before(:example) do
+      stub_request(:put, upload_url)
+        .with(
+          headers: {
+            'Content-Length' => '0',
+            'Content-Range' => 'bytes */33'
+          }
+        )
+        .to_return(status: [499])
+    end
+
+    it 'should cancel a resumable upload' do
+      command.options.upload_chunk_size = 11
+      command.upload_id = upload_id
+      command.delete_upload = true
+      command.execute(client)
+      expect(a_request(:delete, upload_url)).to have_been_made
+    end
+
+    it 'should not call resumable upload when upload is cancelled' do
+      command.options.upload_chunk_size = 11
+      command.upload_id = upload_id
+      command.execute(client)
+      expect(a_request(:put, upload_url)
+        .with(body: 'Hello world')).to have_not_been_made
+    end
+  end
+
   context('with chunking disabled') do
     let!(:file) { StringIO.new("Hello world")}
     include_examples 'should upload'