Class: HealthyPools::TimedStack

Inherits:
Object
  • Object
show all
Defined in:
lib/healthy_pools/timed_stack.rb

Overview

Examples:

ts = TimedStack.new(1) { MyConnection.new }

# fetch a connection
conn = ts.pop

# return a connection
ts.push conn

conn = ts.pop
ts.pop timeout: 5
#=> raises Timeout::Error after 5 seconds

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(size = 0, health_check: nil, &block) ⇒ TimedStack

Creates a new pool with size connections that are created from the given block.



37
38
39
40
41
42
43
44
45
46
 
# File 'lib/healthy_pools/timed_stack.rb', line 37

def initialize(size = 0, health_check: nil, &block)
  @create_block = block
  @health_check = health_check
  @created = 0
  @que = []
  @max = size
  @mutex = Mutex.new
  @resource = ConditionVariable.new
  @shutdown_block = nil
end

Instance Attribute Details

#max ⇒ Object (readonly)

Returns the value of attribute max.



31
32
33
 
# File 'lib/healthy_pools/timed_stack.rb', line 31

def max
  @max
end

Instance Method Details

#delete(conn) ⇒ Object

Removes conn from the stack. This is useful if connections die or otherwise become stale.



52
53
54
55
56
57
58
59
60
 
# File 'lib/healthy_pools/timed_stack.rb', line 52

def delete(conn)
  if @mutex.owned?
    @created -= 1
  else
    @mutex.synchronize do
      @created -= 1
    end
  end
end

#empty? ⇒ Boolean

Returns true if there are no available connections.

Returns:

  • (Boolean) 


136
137
138
 
# File 'lib/healthy_pools/timed_stack.rb', line 136

def empty?
  (@created - @que.length) >= @max
end

#length ⇒ Object

The number of connections available on the stack.



143
144
145
 
# File 'lib/healthy_pools/timed_stack.rb', line 143

def length
  @max - @created + @que.length
end

#pop(timeout = 0.5, options = {}) ⇒ Object

Retrieves a connection from the stack. If a connection is available it is immediately returned. If no connection is available within the given timeout a Timeout::Error is raised.

:timeout is the only checked entry in options and is preferred over the timeout argument (which will be removed in a future release). Other options may be used by subclasses that extend TimedStack.



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
 
# File 'lib/healthy_pools/timed_stack.rb', line 88

def pop(timeout = 0.5, options = {})
  options, timeout = timeout, 0.5 if Hash === timeout
  timeout = options.fetch :timeout, timeout

  deadline = HealthyPools.monotonic_time + timeout
  @mutex.synchronize do
    loop do
      raise HealthyPools::PoolShuttingDownError if @shutdown_block
      while connection_stored?(options)
        conn = fetch_connection(options)
        begin
          if @health_check.nil? || @health_check.call(conn)
            return conn
          end
        rescue
        end
        # Health check has failed, delete conn and retry.
        delete(conn)
      end

      connection = try_create(options)
      return connection if connection

      to_wait = deadline - HealthyPools.monotonic_time
      raise Timeout::Error, "Waited #{timeout} sec" if to_wait <= 0
      @resource.wait(@mutex, to_wait)
    end
  end
end

#push(obj, options = {}) ⇒ Object Also known as: <<

Returns obj to the stack. options is ignored in TimedStack but may be used by subclasses that extend TimedStack.



66
67
68
69
70
71
72
73
74
75
76
 
# File 'lib/healthy_pools/timed_stack.rb', line 66

def push(obj, options = {})
  @mutex.synchronize do
    if @shutdown_block
      @shutdown_block.call(obj)
    else
      store_connection obj, options
    end

    @resource.broadcast
  end
end

#shutdown(&block) ⇒ Object

Shuts down the TimedStack which prevents connections from being checked out. The block is called once for each connection on the stack.

Raises:

  • (ArgumentError) 


122
123
124
125
126
127
128
129
130
131
 
# File 'lib/healthy_pools/timed_stack.rb', line 122

def shutdown(&block)
  raise ArgumentError, "shutdown must receive a block" unless block_given?

  @mutex.synchronize do
    @shutdown_block = block
    @resource.broadcast

    shutdown_connections
  end
end