Attempt at piping HTTP GET to HTTP PUT for direct to S3 transfer

http_get_stream.rb

require 'httparty'
require 'em-http-request'

class HttpGetStream

  def initialize(http_url, listener = nil)
    @http_url = http_url
    @buffer = ''
    self.listener = listener
  end

  def listener=(listener)
    @listener = listener
    listen unless @listener.nil?
  end
  
  def content_length
    response = HTTParty.head(@http_url)
    response['Content-length']
  end

  private

  def listen
    http = EventMachine::HttpRequest.new(@http_url).get
    http.stream do |chunk|
      @buffer += chunk
      process_buffer
    end
  end

  def process_buffer
    @listener.send_data @buffer.slice!(/.*/)
  end

end

http_to_s3_stream.rb

require 'em-http-request'

class HttpToS3Stream

  def initialize(http_url, s3_bucket, s3_key, s3_access_key_id, s3_secret_access_key)
    @http_url = http_url
    @s3_bucket = s3_bucket
    @s3_key = s3_key
    @s3_access_key_id = s3_access_key_id
    @s3_secret_access_key = s3_secret_access_key
    go
  end

  private

  def go
    EM.run {
      # initialize get stream, without listener does not start request
      @get_stream = HttpGetStream.new(@http_url)

      # initialize put stream, send content length, request starts
      @put_stream = S3PutStream.new(@s3_bucket, @s3_key, @s3_access_key_id, @s3_secret_access_key, @get_stream.content_length)

      # set listener on get stream, starts request, pipes data to put stream
      @get_stream.listener = @put_stream
    }
  end

end

run.rb

# stream src_url to dest_bucket/dest_key
HttpToS3Stream.new(src_url, dest_bucket, dest_key, aws_access_key_id, aws_secret_access_key)

s3_put_stream.rb

require 'em-http-request'

class S3PutStream

  def initialize(s3_bucket, s3_key, s3_access_key_id, s3_secret_access_key, content_length = nil)
    @s3_bucket = s3_bucket
    @s3_key = s3_key
    @s3_access_key_id = s3_access_key_id
    @s3_secret_access_key = s3_secret_access_key
    @content_length = content_length
    @bytes_sent = 0
    listen
  end

  def send_data(data)
    @bytes_sent += data.length
    @http.on_body_data data
  end

  private

  def listen
    raise 'ContentLengthRequired' if @content_length.nil?
    @http = EventMachine::HttpRequest.new(put_url).put(
      :head => {
        'Content-Length' => @content_length,
        'Date' => Time.now.getutc,
        'Authorization' => auth_key
      }
    )
    @http.errback { |error| puts "error: #{error}" }
  end

  def put_url
    "http://#{@s3_bucket}.s3.amazonaws.com/#{@s3_key}"
  end

  def auth_key
    "#{@s3_access_key_id}:#{@s3_secret_access_key}"
  end

end