Class: HTTPX::Connection::HTTP1

Inherits:
Object
  • Object
show all
Includes:
HTTPX::Callbacks, Loggable
Defined in:
lib/httpx/connection/http1.rb

Direct Known Subclasses

Plugins::Proxy::HTTP::ProxyParser

Constant Summary collapse

MAX_REQUESTS = 
100
CRLF = 
"\r\n"

Constants included from Loggable

Loggable::COLORS

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from Loggable

#log, #log_exception

Methods included from HTTPX::Callbacks

#emit, #on, #once

Constructor Details

#initialize(buffer, options) ⇒ HTTP1

Returns a new instance of HTTP1.



15
16
17
18
19
20
21
22
23
24
 
# File 'lib/httpx/connection/http1.rb', line 15

def initialize(buffer, options)
  @options = Options.new(options)
  @max_concurrent_requests = @options.max_concurrent_requests || MAX_REQUESTS
  @max_requests = @options.max_requests || MAX_REQUESTS
  @parser = Parser::HTTP1.new(self)
  @buffer = buffer
  @version = [1, 1]
  @pending = []
  @requests = []
end

Instance Attribute Details

#pending ⇒ Object (readonly)

Returns the value of attribute pending.



13
14
15
 
# File 'lib/httpx/connection/http1.rb', line 13

def pending
  @pending
end

Instance Method Details

#<<(data) ⇒ Object 



59
60
61
 
# File 'lib/httpx/connection/http1.rb', line 59

def <<(data)
  @parser << data
end

#close ⇒ Object 



44
45
46
47
 
# File 'lib/httpx/connection/http1.rb', line 44

def close
  reset
  emit(:close, true)
end

#consume ⇒ Object 



75
76
77
78
79
80
81
82
83
 
# File 'lib/httpx/connection/http1.rb', line 75

def consume
  requests_limit = [@max_concurrent_requests, @max_requests, @requests.size].min
  @requests.each_with_index do |request, idx|
    break if idx >= requests_limit
    next if request.state == :done

    handle(request)
  end
end

#dispatch ⇒ Object 



137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
 
# File 'lib/httpx/connection/http1.rb', line 137

def dispatch
  if @request.expects?
    @parser.reset!
    return handle(@request)
  end

  request = @request
  @request = nil
  @requests.shift
  response = request.response
  emit(:response, request, response)

  if @parser.upgrade?
    response << @parser.upgrade_data
    throw(:called)
  end

  @parser.reset!
  @max_requests -= 1
  manage_connection(response)
  send(@pending.shift) unless @pending.empty?
end

#empty? ⇒ Boolean

Returns:

  • (Boolean) 


53
54
55
56
57
 
# File 'lib/httpx/connection/http1.rb', line 53

def empty?
  # this means that for every request there's an available
  # partial response, so there are no in-flight requests waiting.
  @requests.empty? || @requests.all? { |request| !request.response.nil? }
end

#exhausted? ⇒ Boolean

Returns:

  • (Boolean) 


49
50
51
 
# File 'lib/httpx/connection/http1.rb', line 49

def exhausted?
  !@max_requests.positive?
end

#handle_error(ex) ⇒ Object 



160
161
162
163
164
165
166
167
168
169
170
171
 
# File 'lib/httpx/connection/http1.rb', line 160

def handle_error(ex)
  if @pipelining
    disable
  else
    @requests.each do |request|
      emit(:error, request, ex)
    end
    @pending.each do |request|
      emit(:error, request, ex)
    end
  end
end

#interests ⇒ Object 



26
27
28
29
30
31
32
33
34
35
36
37
 
# File 'lib/httpx/connection/http1.rb', line 26

def interests
  # this means we're processing incoming response already
  return :r if @request

  return if @requests.empty?

  request = @requests.first

  return :w if request.interests == :w || !@buffer.empty?

  :r
end

#on_complete ⇒ Object 



130
131
132
133
134
135
 
# File 'lib/httpx/connection/http1.rb', line 130

def on_complete
  return unless @request

  log(level: 2) { "parsing complete" }
  dispatch
end

#on_data(chunk) ⇒ Object 



120
121
122
123
124
125
126
127
128
 
# File 'lib/httpx/connection/http1.rb', line 120

def on_data(chunk)
  return unless @request

  log(color: :green) { "-> DATA: #{chunk.bytesize} bytes..." }
  log(level: 2, color: :green) { "-> #{chunk.inspect}" }
  response = @request.response

  response << chunk
end

#on_headers(h) ⇒ Object 



93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/httpx/connection/http1.rb', line 93

def on_headers(h)
  @request = @requests.first
  return if @request.response

  log(level: 2) { "headers received" }
  headers = @request.options.headers_class.new(h)
  response = @request.options.response_class.new(@request,
                                                 @parser.status_code,
                                                 @parser.http_version.join("."),
                                                 headers)
  log(color: :yellow) { "-> HEADLINE: #{response.status} HTTP/#{@parser.http_version.join(".")}" }
  log(color: :yellow) { response.headers.each.map { |f, v| "-> HEADER: #{f}: #{v}" }.join("\n") }

  @request.response = response
  on_complete if response.complete?
end

#on_start ⇒ Object

HTTP Parser callbacks

must be public methods, or else they won’t be reachable



89
90
91
 
# File 'lib/httpx/connection/http1.rb', line 89

def on_start
  log(level: 2) { "parsing begins" }
end

#on_trailers(h) ⇒ Object 



110
111
112
113
114
115
116
117
118
 
# File 'lib/httpx/connection/http1.rb', line 110

def on_trailers(h)
  return unless @request

  response = @request.response
  log(level: 2) { "trailer headers received" }

  log(color: :yellow) { h.each.map { |f, v| "-> HEADER: #{f}: #{v}" }.join("\n") }
  response.merge_headers(h)
end

#reset ⇒ Object 



39
40
41
42
 
# File 'lib/httpx/connection/http1.rb', line 39

def reset
  @max_requests = @options.max_requests || MAX_REQUESTS
  @parser.reset!
end

#send(request) ⇒ Object 



63
64
65
66
67
68
69
70
71
72
73
 
# File 'lib/httpx/connection/http1.rb', line 63

def send(request)
  unless @max_requests.positive?
    @pending << request
    return
  end

  return if @requests.include?(request)

  @requests << request
  @pipelining = true if @requests.size > 1
end