Class: ScalarGalaxy::Internal::Transport::PooledNetRequester Private

Inherits:
Object
  • Object
show all
Extended by:
Util::SorbetRuntimeSupport
Defined in:
lib/galaxy-ruby/internal/transport/pooled_net_requester.rb,
sig/galaxy-ruby/internal/transport/pooled_net_requester.rbs

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.

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

Returns:

  • (30) 
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.

Returns:

  • (Integer) 
[Etc.nprocessors, 99].max
URI =

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.

Returns:

  • (:Generic url,) 

      
    

  







  
    

      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)  ⇒ 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)
      
      
    
    • 
  
    
  • 
      
        size:
      
      
        (Integer)
      
      
        (defaults to: self.class::DEFAULT_MAX_CONNECTIONS)
      
      
    
    • 
  






  

    
      


203
204
205
206
207
208

    		
    
      
# File 'lib/galaxy-ruby/internal/transport/pooled_net_requester.rb', line 203

def initialize(size: self.class::DEFAULT_MAX_CONNECTIONS)
  @mutex = Mutex.new
  @size = size
  @cert_store = OpenSSL::X509::Store.new.tap(&:set_default_paths)
  @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))
      
      
      
    
    • 
  





  

    
      


67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94

    		
    
      
# File 'lib/galaxy-ruby/internal/transport/pooled_net_requester.rb', line 67

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

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

    		
  





    
      

  

  
    .calibrate_socket_timeout(conn, deadline)  ⇒ void 
  

  

  


  

    

  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.



This method returns an undefined value.


  





  
Parameters:


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






  

    
      


48
49
50
51

    		
    
      
# File 'lib/galaxy-ruby/internal/transport/pooled_net_requester.rb', line 48

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

    		
  





    
      

  

  
    .connect(cert_store:, 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)
      
      
      
    
    • 
  
    
  • 
      
        url
      
      
        (URI::Generic)
      
      
      
    
    • 
  




Returns:


    
  
    
  • 
      
      
        (Net::HTTP)
      
      
      
    
    • 
  





  

    
      


23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42

    		
    
      
# File 'lib/galaxy-ruby/internal/transport/pooled_net_requester.rb', line 23

def connect(cert_store:, 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

      (_1.cert_store = cert_store) if _1.use_ssl?
    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>))
      
      
      
    
    • 
  





  

    
      


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
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198

    		
    
      
# File 'lib/galaxy-ruby/internal/transport/pooled_net_requester.rb', line 133

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

  req = nil
  finished = false

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

      with_pool(url, 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) { self.class.calibrate_socket_timeout(conn, deadline) }

            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 ScalarGalaxy::Errors::APITimeoutError.new(url: url, request: req)
    rescue StandardError
      raise ScalarGalaxy::Errors::APIConnectionError.new(url: url, request: req)
    end
  # rubocop:enable Metrics/BlockLength

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