Class: Subscriber::HTTPChunkedClient

Inherits:

LongPollClient

• Object
• Client
• LongPollClient
• Subscriber::HTTPChunkedClient

Includes:

Celluloid::IO

Defined in:

lib/nchan_tools/pubsub.rb

Instance Attribute Summary

Attributes inherited from LongPollClient

#timeout

Attributes inherited from Client

#concurrency

Class Method Summary

• .aliases ⇒ Object

Instance Method Summary

• #new_bundle(uri, opt) ⇒ Object
• #provides_msgid? ⇒ Boolean
• #run(*args) ⇒ Object
• #setup_bundle(b) ⇒ Object

Methods inherited from LongPollClient

#close, #error, #initialize, #listen, #request_code_ok, #stop

Methods inherited from Client

#error, #handle_bundle_error, inherited, #initialize, lookup, #poke, #stop, unique_aliases

Constructor Details

This class inherits a constructor from Subscriber::LongPollClient

Class Method Details

.aliases ⇒ Object

# File 'lib/nchan_tools/pubsub.rb', line 1441

def self.aliases
  [:chunked]
end

Instance Method Details

#new_bundle(uri, opt) ⇒ Object

# File 'lib/nchan_tools/pubsub.rb', line 1445

def new_bundle(uri, opt)
   opt[:accept]="*/*"
   opt[:headers]=(opt[:headers] or {}).merge({"TE" => "Chunked"})
  super
end

#provides_msgid? ⇒ Boolean

Returns:

• (Boolean)

# File 'lib/nchan_tools/pubsub.rb', line 1428

def provides_msgid?
  false
end

#run(*args) ⇒ Object

# File 'lib/nchan_tools/pubsub.rb', line 1432

def run(*args)
  if @http2
    @subscriber.on_failure error(0, "Chunked transfer is not allowed in HTTP/2")
    @connected = 0
    return
  end
  super
end

#setup_bundle(b) ⇒ Object

# File 'lib/nchan_tools/pubsub.rb', line 1451

def setup_bundle(b)
  super
  b.body_buf = nil
  b.on_headers do |code, headers|
    if code == 200
      if headers["Transfer-Encoding"] != "chunked"
        @subscriber.on_failure error(0, "Transfer-Encoding should be 'chunked', was '#{headers["Transfer-Encoding"]}'.", b)
        close b
      else
        @notready -= 1
        @cooked_ready.signal true if @notready == 0
        b.connected= true
      end
    else
      b.buffer_body!
      b.stop_after_headers = false
    end
  end
  
  b.stop_after_headers = true
  @inchunk = false
  @chunksize = 0
  @repeat = true
  @chunkbuf = ""
  b.on_chunk do |chunk|
    #puts "yeah"
    @chunkbuf << chunk
    @repeat = true
    while @repeat
      @repeat = false
      if !@inchunk && @chunkbuf.slice!(/^([a-fA-F0-9]+)\r\n/m)
        @chunksize = $~[1].to_i(16)
        @inchunk = true
      end
      
      if @inchunk
        if @chunkbuf.length >= @chunksize + 2
          msgbody = @chunkbuf.slice!(0...@chunksize)
          @chunkbuf.slice!(/^\r\n/m)
          @timer.reset if @timer
          unless @nomsg
            msg=Message.new msgbody, nil, nil
          else
            msg=msgbody
          end
          if @subscriber.on_message(msg, b) == false
            @subscriber.finished+=1
            close b
          end
          @repeat = true if @chunkbuf.length > 0
          @inchunk = false
          @chunksize = 0
        end
      end
    end
  end
  
  b.on_response do |code, headers, body|
    if code != 200
      @subscriber.on_failure(error(code, "", b))
    else
      @subscriber.on_failure error(410, "Server Closed Connection", b)
    end
    close b
  end
  
  b
end