Class: DockerEngineRuby::Internal::Transport::PooledNetRequester Private

Inherits:
Object
  • Object
show all
Extended by:
Util::SorbetRuntimeSupport
Defined in:
lib/docker_engine_ruby/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.

Defined Under Namespace

Classes: UnixSocketHTTP

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.

from the golang stdlib github.com/golang/go/blob/c8eced8580028328fde7c03cbfcb720ce15b2358/src/net/http/transport.go#L49

30
DEFAULT_MAX_CONNECTIONS =

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. 

[Etc.nprocessors, 99].max

Class Method Summary collapse

Instance Method Summary collapse

Methods included from Util::SorbetRuntimeSupport

const_missing, define_sorbet_constant!, sorbet_constant_defined?, to_sorbet_type, to_sorbet_type

Constructor Details

#initialize(size: self.class::DEFAULT_MAX_CONNECTIONS, unix_socket_path: nil, tls_verify_peer: true, tls_ca_cert_path: nil, tls_client_cert_path: nil, tls_client_key_path: nil) ⇒ 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: self.class::DEFAULT_MAX_CONNECTIONS)
  • tls_ca_cert_path (String, nil) (defaults to: nil)
  • tls_client_cert_path (String, nil) (defaults to: nil)
  • tls_client_key_path (String, nil) (defaults to: nil)
  • unix_socket_path (String, nil) (defaults to: nil)
  • tls_verify_peer (Boolean) (defaults to: true) 


243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
 
# File 'lib/docker_engine_ruby/internal/transport/pooled_net_requester.rb', line 243

def initialize(
  size: self.class::DEFAULT_MAX_CONNECTIONS,
  unix_socket_path: nil,
  tls_verify_peer: true,
  tls_ca_cert_path: nil,
  tls_client_cert_path: nil,
  tls_client_key_path: nil
)
  @mutex = Mutex.new
  @size = size
  @default_unix_socket_path = unix_socket_path
  @tls_verify_peer = tls_verify_peer
  @cert_store = OpenSSL::X509::Store.new.tap(&:set_default_paths)
  @cert_store.add_file(tls_ca_cert_path) if tls_ca_cert_path

  if tls_client_cert_path || tls_client_key_path
    if tls_client_cert_path.nil? || tls_client_key_path.nil?
      raise ArgumentError.new(
        "Both tls_client_cert_path and tls_client_key_path must be provided together."
      )
    end

    @tls_cert = OpenSSL::X509::Certificate.new(File.read(tls_client_cert_path))
    @tls_key = OpenSSL::PKey.read(File.read(tls_client_key_path))
  else
    @tls_cert = nil
    @tls_key = nil
  end

  @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)) 


97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
 
# File 'lib/docker_engine_ruby/internal/transport/pooled_net_requester.rb', line 97

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
    req["content-length"] ||= 0 unless req["transfer-encoding"]
  in String
    req["content-length"] ||= body.bytesize.to_s unless req["transfer-encoding"]
    req.body_stream = DockerEngineRuby::Internal::Util::ReadIOAdapter.new(body, &blk)
  in StringIO
    req["content-length"] ||= body.size.to_s unless req["transfer-encoding"]
    req.body_stream = DockerEngineRuby::Internal::Util::ReadIOAdapter.new(body, &blk)
  in Pathname | IO | Enumerator
    req["transfer-encoding"] ||= "chunked" unless req["content-length"]
    req.body_stream = DockerEngineRuby::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) 


78
79
80
81
 
# File 'lib/docker_engine_ruby/internal/transport/pooled_net_requester.rb', line 78

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

.connect(cert_store:, tls_cert:, tls_key:, unix_socket_path:, tls_verify_peer:, 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:

  • cert_store (OpenSSL::X509::Store)
  • tls_cert (OpenSSL::X509::Certificate, nil)
  • tls_key (OpenSSL::PKey::PKey, nil)
  • unix_socket_path (String, nil)
  • tls_verify_peer (Boolean)
  • url (URI::Generic)

Returns:

  • (Net::HTTP) 


44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/docker_engine_ruby/internal/transport/pooled_net_requester.rb', line 44

def connect(cert_store:, tls_cert:, tls_key:, unix_socket_path:, tls_verify_peer:, url:)
  if unix_socket_path
    return UnixSocketHTTP.new(unix_socket_path).tap do
      _1.use_ssl = false
      _1.max_retries = 0
    end
  end
  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

    if _1.use_ssl?
      _1.cert_store = cert_store
      _1.cert = tls_cert if tls_cert
      _1.key = tls_key if tls_key
      _1.verify_mode = tls_verify_peer ? OpenSSL::SSL::VERIFY_PEER : OpenSSL::SSL::VERIFY_NONE
    end
  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>)) 


170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/docker_engine_ruby/internal/transport/pooled_net_requester.rb', line 170

def execute(request)
  url, deadline = request.fetch_values(:url, :deadline)
  unix_socket_path = request.fetch(:unix_socket_path, @default_unix_socket_path)

  req = nil
  finished = false

  # rubocop:disable Metrics/BlockLength
  enum = Enumerator.new do |y|
    next if finished

    with_pool(url, unix_socket_path: unix_socket_path, deadline: deadline) do |conn|
      eof = false
      closing = nil
      ::Thread.handle_interrupt(Object => :never) do
        ::Thread.handle_interrupt(Object => :immediate) do
          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)
          ::Kernel.catch(:jump) do
            conn.request(req) do |rsp|
              y << [req, rsp]
              ::Kernel.throw(:jump) if finished

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

                self.class.calibrate_socket_timeout(conn, deadline)
              end
              eof = true
            end
          end
        end
      ensure
        begin
          conn.finish if !eof && conn&.started?
        ensure
          closing&.call
        end
      end
    end
  rescue Timeout::Error
    raise DockerEngineRuby::Errors::APITimeoutError.new(url: url, request: req)
  rescue StandardError
    raise DockerEngineRuby::Errors::APIConnectionError.new(url: url, request: req)
  end
  # rubocop:enable Metrics/BlockLength

  _, response = enum.next
  body = DockerEngineRuby::Internal::Util.fused_enum(enum, external: true) do
    finished = true
    loop { enum.next }
  end
  [Integer(response.code), response, body]
end