Module: Gearman::Evented::Reactor

Includes:
EM::Deferrable
Included in:
ClientReactor, WorkerReactor
Defined in:
lib/gearman/evented/reactor.rb

Class Method Summary collapse

Instance Method Summary collapse

Class Method Details

.connect(host, port, reactor, opts = {}) ⇒ Object 



15
16
17
18
19
20
21
22
23
 
# File 'lib/gearman/evented/reactor.rb', line 15

def self.connect(host, port, reactor, opts = {})
  EM.connect(host, (port || 4730), reactor) do |c|
    c.instance_eval do
      @host = host
      @port = port || 4730
      @opts = opts
    end
  end
end

.included(mod) ⇒ Object 



7
8
9
10
11
12
13
 
# File 'lib/gearman/evented/reactor.rb', line 7

def self.included(mod)
  mod.instance_eval do
    def connect(host, port, opts = {})
      Gearman::Evented::Reactor.connect(host, port, self, opts)
    end
  end
end

Instance Method Details

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


25
26
27
 
# File 'lib/gearman/evented/reactor.rb', line 25

def connected?
  @connected
end

#connection_completed ⇒ Object 



33
34
35
36
37
38
39
 
# File 'lib/gearman/evented/reactor.rb', line 33

def connection_completed
  log "connected to #{@host}:#{@port}"
  @connected = true
  @reconnecting = false
  @reconnect = true
  succeed
end

#disconnect ⇒ Object 



48
49
50
51
52
 
# File 'lib/gearman/evented/reactor.rb', line 48

def disconnect
  log "force disconnect from #{@host}:#{@port}"
  @reconnect = false
  close_connection_after_writing
end

#log(msg, force = false) ⇒ Object 



76
77
78
 
# File 'lib/gearman/evented/reactor.rb', line 76

def log(msg, force = false)
  Gearman::Util.log(msg, force)
end

#reconnect(force = false) ⇒ Object 



54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
 
# File 'lib/gearman/evented/reactor.rb', line 54

def reconnect(force = false)
  if @reconnecting
    EM.add_timer(@opts[:reconnect_sec] || 30) { reconnect }
    return
  elsif !@reconnect && !force
    log "forced disconnect, aborting reconnect attempt"
    @reconnect = true
    return
  else
    @reconnecting = true
    @deferred_status = nil
  end

  log "reconnecting to #{@host}:#{@port}"
  EM.reconnect(@host, @port.to_i, self)
end

#send(command, data = nil) ⇒ Object 



71
72
73
74
 
# File 'lib/gearman/evented/reactor.rb', line 71

def send(command, data = nil)
  log "send #{command} #{data.inspect} (#{server})"
  send_data(Gearman::Protocol.encode_request(command, data))
end

#server ⇒ Object 



29
30
31
 
# File 'lib/gearman/evented/reactor.rb', line 29

def server
  @hostport ||= [ @host, @port ].join(":")
end

#to_s ⇒ Object 



80
81
82
 
# File 'lib/gearman/evented/reactor.rb', line 80

def to_s
  "#{@host}:#{@port}"
end

#unbind ⇒ Object 



41
42
43
44
45
46
 
# File 'lib/gearman/evented/reactor.rb', line 41

def unbind
  log "disconnected from #{@host}:#{@port}"
  @connected = false
  EM.next_tick { reconnect } unless !@reconnect
  @reconnect = true
end