Class: Pangea::Internal::Transport::PooledNetRequester Private

Inherits:
Object
  • Object
show all
Defined in:
lib/pangea/internal/transport/pooled_net_requester.rb

This class is part of a private API. You should avoid using this class if possible, as it may be removed or be changed in the future.

Constant Summary collapse

KEEP_ALIVE_TIMEOUT =

This constant is part of a private API. You should avoid using this constant if possible, as it may be removed or be changed in the future. 

30

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size: Etc.nprocessors) ⇒ PooledNetRequester

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Returns a new instance of PooledNetRequester.

Parameters:

  • size (Integer) (defaults to: Etc.nprocessors) 


183
184
185
186
187
 
# File 'lib/pangea/internal/transport/pooled_net_requester.rb', line 183

def initialize(size: Etc.nprocessors)
  @mutex = Mutex.new
  @size = size
  @pools = {}
end

Class Method Details

.build_request(request, &blk) {|| ... } ⇒ Array(Net::HTTPGenericRequest, Proc)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • request (Hash{Symbol=>Object}) —

    .

    @option request [Symbol] :method

    @option request [URI::Generic] :url

    @option request [HashString=>String] :headers

  • blk (Proc)

Yield Parameters:

  • (String)

Returns:

  • (Array(Net::HTTPGenericRequest, Proc)) 


56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
 
# File 'lib/pangea/internal/transport/pooled_net_requester.rb', line 56

def build_request(request, &blk)
  method, url, headers, body = request.fetch_values(:method, :url, :headers, :body)
  req = Net::HTTPGenericRequest.new(
    method.to_s.upcase,
    !body.nil?,
    method != :head,
    URI(url.to_s) # ensure we construct a URI class of the right scheme
  )

  headers.each { req[_1] = _2 }

  case body
  in nil
    nil
  in String
    req["content-length"] ||= body.bytesize.to_s unless req["transfer-encoding"]
    req.body_stream = Pangea::Internal::Util::ReadIOAdapter.new(body, &blk)
  in StringIO
    req["content-length"] ||= body.size.to_s unless req["transfer-encoding"]
    req.body_stream = Pangea::Internal::Util::ReadIOAdapter.new(body, &blk)
  in Pathname | IO | Enumerator
    req["transfer-encoding"] ||= "chunked" unless req["content-length"]
    req.body_stream = Pangea::Internal::Util::ReadIOAdapter.new(body, &blk)
  end

  [req, req.body_stream&.method(:close)]
end

.calibrate_socket_timeout(conn, deadline) ⇒ Object

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • conn (Net::HTTP)
  • deadline (Float) 


37
38
39
40
 
# File 'lib/pangea/internal/transport/pooled_net_requester.rb', line 37

def calibrate_socket_timeout(conn, deadline)
  timeout = deadline - Pangea::Internal::Util.monotonic_secs
  conn.open_timeout = conn.read_timeout = conn.write_timeout = conn.continue_timeout = timeout
end

.connect(url) ⇒ Net::HTTP

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • url (URI::Generic)

Returns:

  • (Net::HTTP) 


16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
 
# File 'lib/pangea/internal/transport/pooled_net_requester.rb', line 16

def connect(url)
  port =
    case [url.port, url.scheme]
    in [Integer, _]
      url.port
    in [nil, "http" | "ws"]
      Net::HTTP.http_default_port
    in [nil, "https" | "wss"]
      Net::HTTP.https_default_port
    end

  Net::HTTP.new(url.host, port).tap do
    _1.use_ssl = %w[https wss].include?(url.scheme)
    _1.max_retries = 0
  end
end

Instance Method Details

#execute(request) ⇒ Array(Integer, Net::HTTPResponse, Enumerable<String>)

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Parameters:

  • request (Hash{Symbol=>Object}) —

    .

    @option request [Symbol] :method

    @option request [URI::Generic] :url

    @option request [HashString=>String] :headers

    @option request [Object] :body

    @option request [Float] :deadline

Returns:

  • (Array(Integer, Net::HTTPResponse, Enumerable<String>)) 


121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
 
# File 'lib/pangea/internal/transport/pooled_net_requester.rb', line 121

def execute(request)
  url, deadline = request.fetch_values(:url, :deadline)

  req = nil
  eof = false
  finished = false
  closing = nil

  # rubocop:disable Metrics/BlockLength
  enum = Enumerator.new do |y|
    with_pool(url, deadline: deadline) do |conn|
      next if finished

      req, closing = self.class.build_request(request) do
        self.class.calibrate_socket_timeout(conn, deadline)
      end

      self.class.calibrate_socket_timeout(conn, deadline)
      unless conn.started?
        conn.keep_alive_timeout = self.class::KEEP_ALIVE_TIMEOUT
        conn.start
      end

      self.class.calibrate_socket_timeout(conn, deadline)
      conn.request(req) do |rsp|
        y << [conn, req, rsp]
        break if finished

        rsp.read_body do |bytes|
          y << bytes.force_encoding(Encoding::BINARY)
          break if finished

          self.class.calibrate_socket_timeout(conn, deadline)
        end
        eof = true
      end
    end
  rescue Timeout::Error
    raise Pangea::Errors::APITimeoutError.new(url: url, request: req)
  rescue StandardError
    raise Pangea::Errors::APIConnectionError.new(url: url, request: req)
  end
  # rubocop:enable Metrics/BlockLength

  conn, _, response = enum.next
  body = Pangea::Internal::Util.fused_enum(enum, external: true) do
    finished = true
    tap do
      enum.next
    rescue StopIteration
      nil
    end
  ensure
    conn.finish if !eof && conn&.started?
    closing&.call
  end
  [Integer(response.code), response, (response.body = body)]
end