support chunked uploads via PUT

This can be useful for streaming to a backend
(this feature needs tests)

3 files changed, 51 insertions, 16 deletions

diff --git a/lib/mogilefs/chunker.rb b/lib/mogilefs/chunker.rb
new file mode 100644
index 0000000..27ff743
--- /dev/null
+++ b/lib/mogilefs/chunker.rb
@@ -0,0 +1,21 @@
+# -*- encoding: binary -*-
+class MogileFS::Chunker
+  CRLF = "\r\n"
+  attr_reader :io
+
+  def initialize(io)
+    @io = io
+  end
+
+  def write(buf)
+    rv = buf.bytesize
+    @io.write("#{rv.to_s(16)}\r\n")
+    @io.write(buf)
+    @io.write(CRLF)
+    rv
+  end
+
+  def flush
+    @io.write("0\r\n\r\n")
+  end
+end
diff --git a/lib/mogilefs/http_file.rb b/lib/mogilefs/http_file.rb
index 261e209..81f6ad4 100644
--- a/lib/mogilefs/http_file.rb
+++ b/lib/mogilefs/http_file.rb
@@ -2,7 +2,7 @@
 # here are internal implementation details, do not use them in your code
 require 'stringio'
 require 'uri'
-require 'mogilefs/backend'
+require 'mogilefs/chunker'
 
 ##
 # HTTPFile wraps up the new file operations for storing files onto an HTTP
@@ -44,31 +44,45 @@ class MogileFS::HTTPFile < StringIO
     @tried = {}
   end
 
+  def request_put(sock, uri, file_size)
+    if file_size
+      sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
+                 "Content-Length: #{file_size}\r\n\r\n")
+      yield sock
+    else
+      sock.write("PUT #{uri.request_uri} HTTP/1.1\r\n" \
+                 "Host: #{uri.host}:#{uri.port}\r\n" \
+                 "Transfer-Encoding: chunked\r\n\r\n")
+      tmp = MogileFS::Chunker.new(sock)
+      rv = yield tmp
+      tmp.flush
+      rv
+    end
+  end
+
   ##
   # Writes an HTTP PUT request to +sock+ to upload the file and
   # returns file size if the socket finished writing
   def upload(devid, uri) # :nodoc:
-    file_size = length
     sock = MogileFS::Socket.tcp(uri.host, uri.port)
 
     if @streaming_io
-      file_size = @streaming_io.length
-      sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
-                 "Content-Length: #{file_size}\r\n\r\n")
-      @streaming_io.call(Proc.new do |data_to_write|
-        sock.write(data_to_write)
-      end)
+      request_put(sock, uri, @streaming_io.length) do |wr|
+        @streaming_io.call(Proc.new do |data_to_write|
+          wr.write(data_to_write)
+        end)
+      end
     elsif @big_io
-      # Don't try to run out of memory
-      File.open(@big_io) do |fp|
-        file_size = fp.stat.size
-        sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
-                   "Content-Length: #{file_size}\r\n\r\n")
-        MogileFS::X.copy_stream(fp, sock)
+      File.open(@big_io) do |rd|
+        stat = rd.stat
+        request_put(sock, uri, stat.file? ? stat.size : nil) do |wr|
+          file_size = MogileFS::X.copy_stream(rd, wr)
+        end
       end
     else
+      file_size = length
       sock.write("PUT #{uri.request_uri} HTTP/1.0\r\n" \
-                 "Content-Length: #{length}\r\n\r\n#{string}")
+                 "Content-Length: #{file_size}\r\n\r\n#{string}")
     end
 
     case line = sock.timed_read(23, "")
diff --git a/lib/mogilefs/mogilefs.rb b/lib/mogilefs/mogilefs.rb
index 812fba6..9e2df0b 100644
--- a/lib/mogilefs/mogilefs.rb
+++ b/lib/mogilefs/mogilefs.rb
@@ -129,7 +129,7 @@ class MogileFS::MogileFS < MogileFS::Client
 
   ##
   # Copies the contents of +file+ into +key+ in class +klass+.  +file+ can be
-  # either a file name or an object that responds to #sysread.
+  # either a file name or an object that responds to #readpartial.
   # Returns size of +file+ stored
 
   def store_file(key, klass, file)