Honor the configured transfer request window

lib/net/sftp/operations/download.rb

@@ -152,6 +152,8 @@ def initialize(sftp, local, remote, options={}, &progress)
       @options = options
       @active = 0
       @properties = options[:properties] || {}
+      requests
+      read_size
 
       self.logger = sftp.logger
 
@@ -222,13 +224,20 @@ def progress; @progress; end
 
       # The number of bytes to read at a time from remote files.
       def read_size
-        options[:read_size] || DEFAULT_READ_SIZE
+        @read_size ||= positive_option(:read_size, DEFAULT_READ_SIZE)
       end
 
       # The number of simultaneou SFTP requests to use to effect the download.
       # Defaults to 16 for recursive downloads.
       def requests
-        options[:requests] || (recursive? ? 16 : 2)
+        @requests ||= positive_option(:requests, recursive? ? 16 : 2)
+      end
+
+      def positive_option(name, default)
+        value = (options[name] || default).to_i
+        raise ArgumentError, ":#{name} must be positive" unless value > 0
+
+        value
       end
 
       # Enqueues as many files and directories from the stack as possible

lib/net/sftp/operations/upload.rb

@@ -153,6 +153,8 @@ def initialize(sftp, local, remote, options={}, &progress) #:nodoc:
 
       @uploads = []
       @recursive = local.respond_to?(:read) ? false : ::File.directory?(local)
+      requests
+      read_size
 
       if recursive?
         @stack = [entries_for(local)]
@@ -164,7 +166,7 @@ def initialize(sftp, local, remote, options={}, &progress) #:nodoc:
           sftp.mkdir(remote) do |response|
             @active -= 1
             raise StatusException.new(response, "mkdir `#{remote}'") unless response.ok?
-            (options[:requests] || RECURSIVE_READERS).to_i.times do
+            requests.times do
               break unless process_next_entry
             end
           end
@@ -225,6 +227,21 @@ def []=(name, value)
       # The progress handler for this instance. Possibly nil.
       def progress; @progress; end
 
+      def requests
+        @requests ||= positive_option(:requests, recursive? ? RECURSIVE_READERS : SINGLE_FILE_READERS)
+      end
+
+      def read_size
+        @read_size ||= positive_option(:read_size, DEFAULT_READ_SIZE)
+      end
+
+      def positive_option(name, default)
+        value = (options[name] || default).to_i
+        raise ArgumentError, ":#{name} must be positive" unless value > 0
+
+        value
+      end
+
       # A simple struct for recording metadata about the file currently being
       # uploaded.
       LiveFile = Struct.new(:local, :remote, :io, :size, :handle)
@@ -326,7 +343,7 @@ def on_open(response)
         write_next_chunk(file)
 
         if !recursive?
-          (options[:requests] || SINGLE_FILE_READERS).to_i.times { write_next_chunk(file) }
+          (requests - 1).times { write_next_chunk(file) }
         end
       end
 
@@ -358,7 +375,7 @@ def write_next_chunk(file)
         else
           @active += 1
           offset = file.io.pos
-          data = file.io.read(options[:read_size] || DEFAULT_READ_SIZE)
+          data = file.io.read(read_size)
           if data.nil?
             update_progress(:close, file)
             request = sftp.close(file.handle, &method(:on_close))

test/test_download.rb

@@ -144,6 +144,17 @@ def test_download_directory_to_buffer_should_fail
     end
   end
 
+  def test_download_should_reject_nonpositive_requests_and_read_size
+    session = stub("sftp", :logger => nil)
+
+    assert_raises(ArgumentError) do
+      Net::SFTP::Operations::Download.new(session, StringIO.new, "/path/to/remote", :requests => 0)
+    end
+    assert_raises(ArgumentError) do
+      Net::SFTP::Operations::Download.new(session, StringIO.new, "/path/to/remote", :read_size => -1)
+    end
+  end
+
   private
 
     def expect_file_transfer(remote, text, opts={})
@@ -286,4 +297,4 @@ def prepare_directory_tree_download(local, remote)
 
       [file1, file2]
     end
-end
+end

test/test_upload.rb

@@ -59,12 +59,12 @@ def test_upload_file_should_read_chunks_of_size(requested_size=nil)
       channel.gets_packet(FXP_HANDLE, :long, 0, :string, "handle")
       channel.sends_packet(FXP_WRITE, :long, 1, :string, "handle", :int64, 0, :string, "a" * size)
       channel.sends_packet(FXP_WRITE, :long, 2, :string, "handle", :int64, size, :string, "b" * size)
-      channel.sends_packet(FXP_WRITE, :long, 3, :string, "handle", :int64, size*2, :string, "c" * size)
       channel.gets_packet(FXP_STATUS, :long, 1, :long, 0)
-      channel.sends_packet(FXP_WRITE, :long, 4, :string, "handle", :int64, size*3, :string, "d" * size)
+      channel.sends_packet(FXP_WRITE, :long, 3, :string, "handle", :int64, size*2, :string, "c" * size)
       channel.gets_packet(FXP_STATUS, :long, 2, :long, 0)
-      channel.sends_packet(FXP_CLOSE, :long, 5, :string, "handle")
+      channel.sends_packet(FXP_WRITE, :long, 4, :string, "handle", :int64, size*3, :string, "d" * size)
       channel.gets_packet(FXP_STATUS, :long, 3, :long, 0)
+      channel.sends_packet(FXP_CLOSE, :long, 5, :string, "handle")
       channel.gets_packet(FXP_STATUS, :long, 4, :long, 0)
       channel.gets_packet(FXP_STATUS, :long, 5, :long, 0)
     end
@@ -90,10 +90,10 @@ def test_upload_file_with_custom_requests_should_start_that_many_writes
       channel.sends_packet(FXP_WRITE, :long, 1, :string, "handle", :int64, 0, :string, "a" * size)
       channel.sends_packet(FXP_WRITE, :long, 2, :string, "handle", :int64, size, :string, "b" * size)
       channel.sends_packet(FXP_WRITE, :long, 3, :string, "handle", :int64, size*2, :string, "c" * size)
-      channel.sends_packet(FXP_WRITE, :long, 4, :string, "handle", :int64, size*3, :string, "d" * size)
       channel.gets_packet(FXP_STATUS, :long, 1, :long, 0)
-      channel.sends_packet(FXP_CLOSE, :long, 5, :string, "handle")
+      channel.sends_packet(FXP_WRITE, :long, 4, :string, "handle", :int64, size*3, :string, "d" * size)
       channel.gets_packet(FXP_STATUS, :long, 2, :long, 0)
+      channel.sends_packet(FXP_CLOSE, :long, 5, :string, "handle")
       channel.gets_packet(FXP_STATUS, :long, 3, :long, 0)
       channel.gets_packet(FXP_STATUS, :long, 4, :long, 0)
       channel.gets_packet(FXP_STATUS, :long, 5, :long, 0)
@@ -156,6 +156,17 @@ def test_upload_io_should_send_io_as_file
     end
   end
 
+  def test_upload_should_reject_nonpositive_requests_and_read_size
+    session = stub("sftp", :logger => nil)
+
+    assert_raises(ArgumentError) do
+      Net::SFTP::Operations::Upload.new(session, StringIO.new, "/path/to/remote", :requests => 0)
+    end
+    assert_raises(ArgumentError) do
+      Net::SFTP::Operations::Upload.new(session, StringIO.new, "/path/to/remote", :read_size => -1)
+    end
+  end
+
   private
 
     def prepare_directory