Class: Anthropic::Internal::Transport::PooledNetRequester Private

Inherits:
Object
  • Object
show all
Extended by:
Util::SorbetRuntimeSupport
Defined in:
lib/anthropic/internal/transport/pooled_net_requester.rb,
sig/anthropic/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

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
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. 


      
    

  







  
    

      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, proxy: 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.


  





  



  

    
      


246
247
248
249
250
251
252

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 246

def initialize(size: self.class::DEFAULT_MAX_CONNECTIONS, proxy: nil)
  @mutex = Mutex.new
  @size = size
  @proxy = self.class.parse_proxy(proxy)
  @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.




  





  

Yield Parameters:


    
  
    
  • 
      
        
      
      
        (String)
      
      
      
    
    • 
  





  

    
      


113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 113

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


  





  



  

    
      


94
95
96
97

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 94

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


    		
  





    
      

  

  
    .connect(cert_store:, url:, proxy: nil)  ⇒ 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.




  





  



  

    
      


24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 24

def connect(cert_store:, url:, proxy: nil)
  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

  conn =
    if proxy.nil?
      # `net/http` falls back to the `http_proxy`/`https_proxy`/`no_proxy` environment variables.
      Net::HTTP.new(url.host, port)
    else
      # `https://` targets are tunnelled through the proxy with CONNECT by `net/http` itself.
      Net::HTTP.new(
        url.host,
        port,
        proxy.hostname,
        proxy.port,
        proxy.user&.then { URI.decode_uri_component(_1) },
        proxy.password&.then { URI.decode_uri_component(_1) }
      )
    end

  conn.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


    		
  





    
      

  

  
    .parse_proxy(proxy)  ⇒ URI::Generic? 
  

  

  




  

    

  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.




  





  

Raises:


    
  
    
  • 
      
      
        (ArgumentError)
      
      
      
    
    • 
  





  

    
      


65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 65

def parse_proxy(proxy)
  return nil if proxy.nil?

  parsed =
    case proxy
    in URI::Generic
      proxy
    in String
      begin
        URI(proxy)
      rescue URI::InvalidURIError
        nil
      end
    else
      nil
    end

  # The offending value is deliberately left out of the message since proxy URLs often embed credentials.
  if parsed.nil? || parsed.scheme&.downcase != "http" || parsed.host.to_s.empty?
    raise ArgumentError.new("Expected proxy to be an http:// URL with a host, for example http://proxy.example.com:8080")
  end

  parsed
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.




  





  



  

    
      


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
234
235
236
237
238
239
240

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 178

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

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


    		
  





    
      

  

  
    #inspect  ⇒ 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.




  





  



  

    
      


257
258
259
260
261

    		
    
      
# File 'lib/anthropic/internal/transport/pooled_net_requester.rb', line 257

def inspect
  # Proxy URLs often embed credentials, so never surface the password.
  proxy = @proxy&.then { _1.password.nil? ? _1 : _1.dup.tap { |uri| uri.password = "REDACTED" } }
  "#<#{self.class.name}:0x#{object_id.to_s(16)} size=#{@size} proxy=#{proxy.inspect}>"
end