lib/unicorn/stream_input.rb

   1 # -*- encoding: binary -*-
   2
   3 # When processing uploads, unicorn may expose a StreamInput object under
   4 # "rack.input" of the Rack environment when
   5 # Unicorn::Configurator#rewindable_input is set to +false+
   6 class Unicorn::StreamInput
   7   # The I/O chunk size (in +bytes+) for I/O operations where
   8   # the size cannot be user-specified when a method is called.
   9   # The default is 16 kilobytes.
  10   @@io_chunk_size = Unicorn::Const::CHUNK_SIZE # :nodoc:
  11
  12   # Initializes a new StreamInput object.  You normally do not have to call
  13   # this unless you are writing an HTTP server.
  14   def initialize(socket, request) # :nodoc:
  15     @chunked = request.content_length.nil?
  16     @socket = socket
  17     @parser = request
  18     @buf = request.buf
  19     @rbuf = ''
  20     @bytes_read = 0
  21     filter_body(@rbuf, @buf) unless @buf.empty?
  22   end
  23
  24   # :call-seq:
  25   #   ios.read([length [, buffer ]]) => string, buffer, or nil
  26   #
  27   # Reads at most length bytes from the I/O stream, or to the end of
  28   # file if length is omitted or is nil. length must be a non-negative
  29   # integer or nil. If the optional buffer argument is present, it
  30   # must reference a String, which will receive the data.
  31   #
  32   # At end of file, it returns nil or '' depend on length.
  33   # ios.read() and ios.read(nil) returns ''.
  34   # ios.read(length [, buffer]) returns nil.
  35   #
  36   # If the Content-Length of the HTTP request is known (as is the common
  37   # case for POST requests), then ios.read(length [, buffer]) will block
  38   # until the specified length is read (or it is the last chunk).
  39   # Otherwise, for uncommon "Transfer-Encoding: chunked" requests,
  40   # ios.read(length [, buffer]) will return immediately if there is
  41   # any data and only block when nothing is available (providing
  42   # IO#readpartial semantics).
  43   def read(length = nil, rv = '')
  44     if length
  45       if length <= @rbuf.size
  46         length < 0 and raise ArgumentError, "negative length #{length} given"
  47         rv.replace(@rbuf.slice!(0, length))
  48       else
  49         to_read = length - @rbuf.size
  50         rv.replace(@rbuf.slice!(0, @rbuf.size))
  51         until to_read == 0 || eof? || (rv.size > 0 && @chunked)
  52           @socket.kgio_read(to_read, @buf) or eof!
  53           filter_body(@rbuf, @buf)
  54           rv << @rbuf
  55           to_read -= @rbuf.size
  56         end
  57         @rbuf.clear
  58       end
  59       rv = nil if rv.empty? && length != 0
  60     else
  61       read_all(rv)
  62     end
  63     rv
  64   end
  65
  66   # :call-seq:
  67   #   ios.gets   => string or nil
  68   #
  69   # Reads the next ``line'' from the I/O stream; lines are separated
  70   # by the global record separator ($/, typically "\n"). A global
  71   # record separator of nil reads the entire unread contents of ios.
  72   # Returns nil if called at the end of file.
  73   # This takes zero arguments for strict Rack::Lint compatibility,
  74   # unlike IO#gets.
  75   def gets
  76     sep = $/
  77     if sep.nil?
  78       read_all(rv = '')
  79       return rv.empty? ? nil : rv
  80     end
  81     re = /\A(.*?#{Regexp.escape(sep)})/
  82
  83     begin
  84       @rbuf.sub!(re, '') and return $1
  85       return @rbuf.empty? ? nil : @rbuf.slice!(0, @rbuf.size) if eof?
  86       @socket.kgio_read(@@io_chunk_size, @buf) or eof!
  87       filter_body(once = '', @buf)
  88       @rbuf << once
  89     end while true
  90   end
  91
  92   # :call-seq:
  93   #   ios.each { |line| block }  => ios
  94   #
  95   # Executes the block for every ``line'' in *ios*, where lines are
  96   # separated by the global record separator ($/, typically "\n").
  97   def each
  98     while line = gets
  99       yield line
 100     end
 101
 102     self # Rack does not specify what the return value is here
 103   end
 104
 105 private
 106
 107   def eof?
 108     if @parser.body_eof?
 109       while @chunked && ! @parser.parse
 110         once = @socket.kgio_read(@@io_chunk_size) or eof!
 111         @buf << once
 112       end
 113       @socket = nil
 114       true
 115     else
 116       false
 117     end
 118   end
 119
 120   def filter_body(dst, src)
 121     rv = @parser.filter_body(dst, src)
 122     @bytes_read += dst.size
 123     rv
 124   end
 125
 126   def read_all(dst)
 127     dst.replace(@rbuf)
 128     @socket or return
 129     until eof?
 130       @socket.kgio_read(@@io_chunk_size, @buf) or eof!
 131       filter_body(@rbuf, @buf)
 132       dst << @rbuf
 133     end
 134   ensure
 135     @rbuf.clear
 136   end
 137
 138   def eof!
 139     # in case client only did a premature shutdown(SHUT_WR)
 140     # we do support clients that shutdown(SHUT_WR) after the
 141     # _entire_ request has been sent, and those will not have
 142     # raised EOFError on us.
 143     @socket.shutdown if @socket
 144   ensure
 145     raise Unicorn::ClientShutdown, "bytes_read=#{@bytes_read}", []
 146   end
 147 end