File: connection_pool.rb

package info (click to toggle)
ruby-aws-sdk 1.67.0-3
  • links: PTS, VCS
  • area: main
  • in suites: forky, sid, trixie
  • size: 6,840 kB
  • sloc: ruby: 28,436; makefile: 7
file content (375 lines) | stat: -rw-r--r-- 12,882 bytes parent folder | download | duplicates (3)
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
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
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
83
84
85
86
87
88
89
90
91
92
93
94
95
96
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
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
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
241
242
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
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
# Copyright 2011-2013 Amazon.com, Inc. or its affiliates. All Rights Reserved.
#
# Licensed under the Apache License, Version 2.0 (the "License"). You
# may not use this file except in compliance with the License. A copy of
# the License is located at
#
#     http://aws.amazon.com/apache2.0/
#
# or in the "license" file accompanying this file. This file is
# distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF
# ANY KIND, either express or implied. See the License for the specific
# language governing permissions and limitations under the License.

require 'net/http'
require 'net/https'
require 'thread'
require 'logger'

module AWS
  module Core
    module Http

      # @attr_reader [URI::HTTP,nil] proxy_uri Returns the configured proxy uri.
      # @attr_reader [Float,nil] http_continue_timeout
      # @attr_reader [Integer,Float] http_idle_timeout
      # @attr_reader [Integer,Float] http_open_timeout
      # @attr_reader [Integer,Float] http_read_timeout
      # @attr_reader [Boolean] http_wire_trace
      # @attr_reader [Logger,nil] logger
      # @attr_reader [Boolean] ssl_verify_peer
      # @attr_reader [String,nil] ssl_ca_file
      # @attr_reader [String,nil] ssl_ca_path
      # @attr_reader [String,nil] ssl_cert_store
      # @api private
      class ConnectionPool

        @pools_mutex = Mutex.new
        @pools = {}

        # @api private
        OPTIONS = [
          :proxy_uri,
          :http_continue_timeout,
          :http_idle_timeout,
          :http_open_timeout,
          :http_read_timeout,
          :http_wire_trace,
          :logger,
          :ssl_verify_peer,
          :ssl_ca_file,
          :ssl_ca_path,
          :ssl_cert_store,
        ]

        OPTIONS.each do |attr_name|
          attr_reader(attr_name)
        end

        alias_method :http_wire_trace?, :http_wire_trace

        alias_method :ssl_verify_peer?, :ssl_verify_peer

        # @api private
        def initialize options = {}
          # user supplied options are filtered by the class .for method
          options.each_pair do |opt_name, opt_value|
            instance_variable_set("@#{opt_name}", opt_value)
          end

          # connection pool
          @pool_mutex = Mutex.new
          @pool = Hash.new do |pool,endpoint|
            pool[endpoint] = []
            pool[endpoint]
          end
        end

        # @return [Hash] a read-only hash of options for this pool.
        def options
          OPTIONS.inject({}) do |options, opt_name|
            options[opt_name] = send(opt_name)
            options
          end.freeze
        end

        # Makes an HTTP request, yielding a Net::HTTPResponse object.
        #
        #   pool.request('http://google.com', Net::HTTP::Get.new('/')) do |resp|
        #     puts resp.code # status code
        #     puts resp.to_h.inspect # dump the headers
        #     puts resp.body
        #   end
        #
        # @param [URI::HTTP,URI::HTTPS,String] endpoint The HTTP(S) endpoint to
        #    connect to (e.g. 'https://domain.com').
        #
        # @param [Net::HTTPRequest] request The request to make.  This can be 
        #   any request object from Net::HTTP (e.g. Net::HTTP::Get,
        #   Net::HTTP::POST, etc).
        #
        # @yieldparam [Net::HTTPResponse] net_http_response
        #
        # @return (see #session_for
        def request endpoint, request, &block
          session_for(endpoint) do |http|
            yield(http.request(request))
          end
        end

        # @param [URI::HTTP,URI::HTTPS,String] endpoint The HTTP(S) endpoint to
        #    connect to (e.g. 'https://domain.com').
        #
        # @yieldparam [Net::HTTPSession] session
        #
        # @return [nil]
        def session_for endpoint, &block
          endpoint = endpoint.to_s
          session = nil

          # attempt to recycle an already open session
          @pool_mutex.synchronize do
            _clean
            session = @pool[endpoint].shift
          end

          begin
            session ||= start_session(endpoint)
            session.read_timeout = http_read_timeout
            session.continue_timeout = http_continue_timeout if
              session.respond_to?(:continue_timeout=)
            yield(session)
          rescue Exception => error
            session.finish if session
            raise error
          else
            # No error raised? Good, check the session into the pool.
            @pool_mutex.synchronize { @pool[endpoint] << session }
          end
          nil
        end

        # @return [Integer] Returns the count of sessions currently in the pool,
        #   not counting those currently in use.
        def size
          @pool_mutex.synchronize do
            size = 0
            @pool.each_pair do |endpoint,sessions|
              size += sessions.size
            end
            size
          end
        end

        # Removes stale http sessions from the pool (that have exceeded
        # the idle timeout).
        # @return [nil]
        def clean!
          @pool_mutex.synchronize { _clean }
          nil
        end

        # Closes and removes removes all sessions from the pool.
        # If empty! is called while there are outstanding requests they may
        # get checked back into the pool, leaving the pool in a non-empty state.
        # @return [nil]
        def empty!
          @pool_mutex.synchronize do
            @pool.each_pair do |endpoint,sessions|
              sessions.each(&:finish)
            end
            @pool.clear
          end
          nil
        end

        class << self

          # Returns a connection pool constructed from the given options.
          # Calling this method twice with the same options will return
          # the same pool.
          #
          # @option options [URI::HTTP,String] :proxy_uri A proxy to send 
          #   requests through.  Formatted like 'http://proxy.com:123'.
          #
          # @option options [Float] :http_continue_timeout (nil) The number of
          #   seconds to wait for a 100-continue response before sending the
          #   request body.  This option has no effect unless the request has
          #   "Expect" header set to "100-continue".  Defaults to `nil` which
          #   disables this behaviour.  This value can safely be set per-request
          #   on the session yeidled by {#session_for}.
          #
          # @option options [Float] :http_idle_timeout (15) The number of
          #   seconds a connection is allowed to sit idble before it is
          #   considered stale.  Stale connections are closed and removed
          #   from the pool before making a request.
          #
          # @option options [Float] :http_open_timeout (15) The number of
          #   seconds to wait when opening a HTTP session before rasing a
          #   `Timeout::Error`.
          #
          # @option options [Integer] :http_read_timeout (60) The default
          #   number of seconds to wait for response data.  This value can
          #   safely be set
          #   per-request on the session yeidled by {#session_for}.
          #
          # @option options [Boolean] :http_wire_trace (false) When `true`, HTTP
          #   debug output will be sent to the `:logger`.
          #
          # @option options [Logger] :logger Where debug output is sent.
          #    Defaults to `nil` when `:http_wire_trace` is `false`.
          #    Defaults to `Logger.new($stdout)` when `:http_wire_trace` is
          #    `true`.
          #
          # @option options [Boolean] :ssl_verify_peer (true) When `true`, SSL
          #   peer certificates are verified when establishing a connection.
          #
          # @option options [String] :ssl_ca_file Full path to the SSL
          #   certificate authority bundle file that should be used when
          #   verifying peer certificates.  If you do not pass
          #   `:ssl_ca_file` or `:ssl_ca_path` the the system default will be
          #   used if available.
          #
          # @option options [String] :ssl_ca_path Full path of the directory
          #   that contains the unbundled SSL certificate authority files#
          #   for verifying peer certificates.  If you do not pass
          #   `:ssl_ca_file` or `:ssl_ca_path` the the system default will
          #   be used if available.
          #
          # @option options [String] :ssl_cert_store
          #
          # @return [ConnectionPool]
          def new options = {}
            options = pool_options(options)
            @pools_mutex.synchronize do
              @pools[options] ||= build(options)
            end
          end

          # Constructs and returns a new connection pool.  This pool is never
          # shared.
          # @option (see new)
          # @return [ConnectionPool]
          def build(options = {})
            pool = allocate
            pool.send(:initialize, pool_options(options))
            pool
          end

          # @return [Array<ConnectionPool>] Returns a list of of the constructed
          #   connection pools.
          def pools
            @pools.values
          end

          private

          # Filters an option hash, merging in default values.
          # @return [Hash]
          def pool_options options
            wire_trace = !!options[:http_wire_trace]
            logger = options[:logger] || Logger.new($stdout) if wire_trace
            verify_peer = options.key?(:ssl_verify_peer) ?
              !!options[:ssl_verify_peer] : true
            {
              :proxy_uri => URI.parse(options[:proxy_uri].to_s),
              :http_continue_timeout => options[:http_continue_timeout],
              :http_open_timeout => options[:http_open_timeout] || 15,
              :http_idle_timeout => options[:http_idle_timeout] || 15,
              :http_read_timeout => options[:http_read_timeout] || 60,
              :http_wire_trace => wire_trace,
              :logger => logger,
              :ssl_verify_peer => verify_peer,
              :ssl_ca_file => options[:ssl_ca_file],
              :ssl_ca_path => options[:ssl_ca_path],
              :ssl_cert_store => options[:ssl_cert_store],
            }
          end

        end

        private

        # Starts and returns a new HTTP(S) session.
        # @param [String] endpoint
        # @return [Net::HTTPSession]
        def start_session endpoint

          endpoint = URI.parse(endpoint)

          args = []
          args << endpoint.host
          args << endpoint.port
          args << proxy_uri.host
          args << proxy_uri.port
          
          if proxy_uri.user
            args << URI::decode(proxy_uri.user)
          else
            args << nil
          end
          
          if proxy_uri.password
            args << URI::decode(proxy_uri.password)
          else 
            args << nil
          end

          http = Net::HTTP.new(*args.compact)
          http.extend(SessionExtensions)
          http.set_debug_output(logger) if http_wire_trace?
          http.open_timeout = http_open_timeout

          if endpoint.scheme == 'https'
            http.use_ssl = true
            if ssl_verify_peer?
              http.verify_mode = OpenSSL::SSL::VERIFY_PEER
              http.ca_file = ssl_ca_file if ssl_ca_file
              http.ca_path = ssl_ca_path if ssl_ca_path
              http.cert_store = ssl_cert_store if ssl_cert_store
            else
              http.verify_mode = OpenSSL::SSL::VERIFY_NONE
            end
          else
            http.use_ssl = false
          end

          http.start
          http
        end

        # Removes stale sessions from the pool.  This method *must* be called
        # @note **Must** be called behind a `@pool_mutex` synchronize block.
        def _clean
          now = Time.now
          @pool.each_pair do |endpoint,sessions|
            sessions.delete_if do |session|
              if
                session.last_used.nil? or
                now - session.last_used > http_idle_timeout
              then
                session.finish
                true
              end
            end
          end
        end

        # Helper methods extended onto Net::HTTPSession objects opend by the
        # connection pool.
        # @api private
        module SessionExtensions

          # Sends the request and tracks that this session has been used.
          def request *args, &block
            @last_used = Time.now
            super(*args, &block)
          end

          # @return [Time,nil]
          def last_used
            @last_used
          end

          # Attempts to close/finish the session without raising an error.
          def finish
            super
          rescue IOError
            nil
          end

        end
      end
    end
  end
end