Class: Gearman::ConnectionPool
- Inherits:
-
Object
- Object
- Gearman::ConnectionPool
- Includes:
- Logging
- Defined in:
- lib/gearman/connection_pool.rb
Constant Summary collapse
- DEFAULT_PORT =
4730
Instance Method Summary collapse
- #add_connection(connection) ⇒ Object
- #add_host_port(host_port) ⇒ Object
- #add_servers(servers) ⇒ Object
- #get_connection(coalesce_key = nil) ⇒ Object
-
#initialize(servers = []) ⇒ ConnectionPool
constructor
A new instance of ConnectionPool.
- #on_connection(&block) ⇒ Object
- #poll_connections(timeout = nil) ⇒ Object
- #with_all_connections(&block) ⇒ Object
Methods included from Logging
Constructor Details
#initialize(servers = []) ⇒ ConnectionPool
Returns a new instance of ConnectionPool.
9 10 11 12 13 14 15 16 17 18 19 20 |
# File 'lib/gearman/connection_pool.rb', line 9 def initialize(servers = []) @bad_servers = [] @coalesce_connections = {} @connection_handler = nil @job_servers = [] @reconnect_seconds = 10 @server_counter = 0 # Round-robin distribution of requests @servers_mutex = Mutex.new add_servers(servers) start_reconnect_thread end |
Instance Method Details
#add_connection(connection) ⇒ Object
22 23 24 25 26 27 28 29 30 |
# File 'lib/gearman/connection_pool.rb', line 22 def add_connection(connection) @servers_mutex.synchronize do if connection.is_healthy? activate_connection(connection) else deactivate_connection(connection) end end end |
#add_host_port(host_port) ⇒ Object
32 33 34 35 36 |
# File 'lib/gearman/connection_pool.rb', line 32 def add_host_port(host_port) host, port = host_port.split(":") connection = Connection.new(host, port.to_i) add_connection(connection) end |
#add_servers(servers) ⇒ Object
38 39 40 41 42 43 44 45 46 47 48 |
# File 'lib/gearman/connection_pool.rb', line 38 def add_servers(servers) if servers.class == String or servers.class == Symbol servers = [ servers.to_s ] end servers = servers.map {|s| s =~ /:/ ? s : "#{s}:#{DEFAULT_PORT}" } servers.each do |host_port| add_host_port(host_port) end end |
#get_connection(coalesce_key = nil) ⇒ Object
50 51 52 53 54 55 56 57 58 |
# File 'lib/gearman/connection_pool.rb', line 50 def get_connection(coalesce_key = nil) @servers_mutex.synchronize do logger.debug "Available job servers: #{@job_servers.inspect}" raise NoJobServersError if @job_servers.empty? @server_counter += 1 @job_servers[@server_counter % @job_servers.size] end end |
#on_connection(&block) ⇒ Object
60 61 62 |
# File 'lib/gearman/connection_pool.rb', line 60 def on_connection(&block) @connection_handler = block end |
#poll_connections(timeout = nil) ⇒ Object
64 65 66 67 68 69 70 71 72 73 |
# File 'lib/gearman/connection_pool.rb', line 64 def poll_connections(timeout = nil) available_sockets = [] @servers_mutex.synchronize do available_sockets.concat @job_servers.collect { |conn| conn.socket } end if available_sockets.size > 0 logger.debug "Polling on #{available_sockets.size} available server(s) with a #{timeout} second timeout" IO::select(available_sockets, nil, nil, timeout) end end |
#with_all_connections(&block) ⇒ Object
75 76 77 78 79 80 81 82 83 84 85 86 |
# File 'lib/gearman/connection_pool.rb', line 75 def with_all_connections(&block) @servers_mutex.synchronize do @job_servers.each do |connection| begin block.call(connection) rescue NetworkError => ex logger.debug "Error with #{connection}, marking as bad" deactivate_connection(connection) end end end end |