Class: Nfe::Http::NetHttp

Inherits:
Object
  • Object
show all
Defined in:
lib/nfe/http/net_http.rb,
sig/nfe/http/net_http.rbs

Overview

Default, zero-dependency HTTP transport built on the Ruby standard library (+net/http+, uri, openssl, zlib, stringio). It satisfies the Transport contract: it returns 4xx/5xx as ordinary Response objects, raises only ApiConnectionError (or its subclass TimeoutError) on network failure, and never follows a redirect or 202.

Connections are pooled per origin ("host:port") and kept alive for TCP/TLS reuse. The pool is guarded by a Mutex and follows a checkout/check-in model: a connection is removed from the idle list while a request is in flight and only returned afterwards. A single instance is therefore safe to share across threads (Rails/Sidekiq/Puma): two in-flight requests never share the same socket, and a connection broken by a network error is closed instead of being returned to the pool.

TLS is always enforced for https origins with OpenSSL::SSL::VERIFY_PEER; verification is never disabled. An optional ca_file provides an escape hatch for a legacy CA store.

Constant Summary collapse

KEEP_ALIVE_TIMEOUT =

Seconds an idle pooled connection is kept open for reuse.

Returns:

  • (Integer) 
30

Instance Method Summary collapse

Constructor Details

#initialize(default_open_timeout: 10, default_read_timeout: 60, ca_file: nil) ⇒ NetHttp

Returns a new instance of NetHttp.

Parameters:

  • default_open_timeout: (Integer) (defaults to: 10)
  • default_read_timeout: (Integer) (defaults to: 60)
  • ca_file: (String, nil) (defaults to: nil) 


33
34
35
36
37
38
39
 
# File 'lib/nfe/http/net_http.rb', line 33

def initialize(default_open_timeout: 10, default_read_timeout: 60, ca_file: nil)
  @default_open_timeout = default_open_timeout
  @default_read_timeout = default_read_timeout
  @ca_file = ca_file
  @pool = {}
  @mutex = Mutex.new
end

Instance Method Details

#accept_encoding_set?(headers) ⇒ Boolean

Parameters:

  • headers (Hash[untyped, untyped])

Returns:

  • (Boolean) 


127
128
129
 
# File 'lib/nfe/http/net_http.rb', line 127

def accept_encoding_set?(headers)
  headers.any? { |name, _| name.to_s.downcase == "accept-encoding" }
end

#apply_timeouts(http, request) ⇒ void

This method returns an undefined value.

Parameters:



107
108
109
110
 
# File 'lib/nfe/http/net_http.rb', line 107

def apply_timeouts(http, request)
  http.open_timeout = request.open_timeout || @default_open_timeout
  http.read_timeout = request.read_timeout || @default_read_timeout
end

#build_connection(uri) ⇒ Object

Parameters:

  • uri (URI::Generic)

Returns:

  • (Object) 


92
93
94
95
96
97
 
# File 'lib/nfe/http/net_http.rb', line 92

def build_connection(uri)
  http = Net::HTTP.new(uri.host.to_s, uri.port)
  http.keep_alive_timeout = KEEP_ALIVE_TIMEOUT
  configure_tls(http, uri)
  http
end

#build_net_request(uri, request) ⇒ Object

Builds the appropriate Net::HTTP request object, copying headers and body from the Request and defaulting Accept-Encoding: gzip when the caller has not set it.

Parameters:

Returns:

  • (Object) 


119
120
121
122
123
124
125
 
# File 'lib/nfe/http/net_http.rb', line 119

def build_net_request(uri, request)
  net_request = request_class(request.method).new(uri)
  request.headers.each { |name, value| net_request[name.to_s] = value }
  net_request["Accept-Encoding"] = "gzip" unless accept_encoding_set?(request.headers)
  net_request.body = request.body unless request.body.nil?
  net_request
end

#build_response(net_response) ⇒ Nfe::Http::Response

Maps a Net::HTTPResponse into an Response: integer status, lowercase headers (multi-value joined with ", "), and a binary body. A gzip body is inflated transparently. Redirects and 202 are not followed.

Parameters:

  • net_response (Object)

Returns:



145
146
147
148
149
150
151
 
# File 'lib/nfe/http/net_http.rb', line 145

def build_response(net_response)
  status = net_response.code.to_i
  headers = normalize_headers(net_response)
  body = (net_response.body || "").dup.force_encoding(Encoding::ASCII_8BIT)
  body, headers = decompress(body, headers)
  Nfe::Http::Response.new(status: status, headers: headers, body: body)
end

#call(request) ⇒ Nfe::Http::Response

Executes request and returns an Response. Raises TimeoutError on open/read timeout and ApiConnectionError on any other network-level failure.

Parameters:

Returns:



44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
 
# File 'lib/nfe/http/net_http.rb', line 44

def call(request)
  uri = URI.parse(request.url)
  key = origin_key(uri)
  http = nil
  begin
    # checkout starts the connection (http.start), which can itself raise a
    # network error (e.g. ECONNREFUSED) — keep it inside the rescue.
    http = checkout(key, uri, request)
    net_response = http.request(build_net_request(uri, request))
    checkin(key, http)
    build_response(net_response)
  rescue Timeout::Error => e
    # Net::OpenTimeout and Net::ReadTimeout both descend from Timeout::Error.
    discard(http) if http
    raise Nfe::TimeoutError, e.message
  rescue SocketError, SystemCallError, OpenSSL::SSL::SSLError,
         Net::HTTPBadResponse, IOError => e
    # IOError covers EOFError; SystemCallError covers all Errno::* errors.
    discard(http) if http
    raise Nfe::ApiConnectionError, e.message
  end
end

#checkin(key, http) ⇒ void

This method returns an undefined value.

Returns a healthy connection to the idle list for reuse.

Parameters:

  • key (String)
  • http (Object) 


81
82
83
 
# File 'lib/nfe/http/net_http.rb', line 81

def checkin(key, http)
  @mutex.synchronize { (@pool[key] ||= []) << http }
end

#checkout(key, uri, request) ⇒ Object

Removes an idle connection for key from the pool (or builds a fresh, started one) and hands it to the caller. While the connection is in flight it is owned exclusively by this call and absent from the pool, so no other thread can use the same socket concurrently. Timeouts are applied per call.

Parameters:

Returns:

  • (Object) 


73
74
75
76
77
78
 
# File 'lib/nfe/http/net_http.rb', line 73

def checkout(key, uri, request)
  http = @mutex.synchronize { (@pool[key] ||= []).pop } || build_connection(uri)
  apply_timeouts(http, request)
  http.start unless http.started?
  http
end

#configure_tls(http, uri) ⇒ void

This method returns an undefined value.

Parameters:

  • http (Object)
  • uri (URI::Generic) 


99
100
101
102
103
104
105
 
# File 'lib/nfe/http/net_http.rb', line 99

def configure_tls(http, uri)
  return unless uri.scheme == "https"

  http.use_ssl = true
  http.verify_mode = OpenSSL::SSL::VERIFY_PEER
  http.ca_file = @ca_file if @ca_file
end

#decompress(body, headers) ⇒ [ String, Hash[String, String] ]

Inflates a gzip body and drops the now-stale content-encoding and content-length headers. On a malformed stream the raw body is kept and a warning is emitted, never raising.

Parameters:

  • body (String)
  • headers (Hash[String, String])

Returns:

  • ([ String, Hash[String, String] ]) 


165
166
167
168
169
170
171
172
173
174
175
176
 
# File 'lib/nfe/http/net_http.rb', line 165

def decompress(body, headers)
  return [body, headers] unless headers["content-encoding"].to_s.downcase.include?("gzip")

  inflated = Zlib::GzipReader.new(StringIO.new(body)).read.to_s.force_encoding(Encoding::ASCII_8BIT)
  cleaned = headers.dup
  cleaned.delete("content-encoding")
  cleaned.delete("content-length")
  [inflated, cleaned]
rescue Zlib::Error => e
  warn("Nfe::Http::NetHttp: failed to inflate gzip response body: #{e.message}")
  [body, headers]
end

#discard(http) ⇒ void

This method returns an undefined value.

Closes a connection broken by a network error so it is never reused.

Parameters:

  • http (Object) 


86
87
88
89
90
 
# File 'lib/nfe/http/net_http.rb', line 86

def discard(http)
  http.finish if http.started?
rescue IOError, SystemCallError
  # Socket already torn down; nothing left to close.
end

#normalize_headers(net_response) ⇒ Hash[String, String]

Parameters:

  • net_response (Object)

Returns:

  • (Hash[String, String]) 


153
154
155
156
157
158
159
160
 
# File 'lib/nfe/http/net_http.rb', line 153

def normalize_headers(net_response)
  headers = {} #: Hash[String, String]
  net_response.each_capitalized do |name, value|
    key = name.downcase
    headers[key] = headers.key?(key) ? "#{headers[key]}, #{value}" : value
  end
  headers
end

#origin_key(uri) ⇒ String

Parameters:

  • uri (URI::Generic)

Returns:

  • (String) 


112
113
114
 
# File 'lib/nfe/http/net_http.rb', line 112

def origin_key(uri)
  "#{uri.host}:#{uri.port}"
end

#request_class(method) ⇒ Object

Parameters:

  • method (Object)

Returns:

  • (Object) 


131
132
133
134
135
136
137
138
139
140
 
# File 'lib/nfe/http/net_http.rb', line 131

def request_class(method)
  case method.to_s.upcase
  when "GET"    then Net::HTTP::Get
  when "POST"   then Net::HTTP::Post
  when "PUT"    then Net::HTTP::Put
  when "DELETE" then Net::HTTP::Delete
  when "HEAD"   then Net::HTTP::Head
  else raise Nfe::ApiConnectionError, "Unsupported HTTP method: #{method}"
  end
end