Class: Smith::Agent

Inherits:
Object
  • Object
show all
Includes:
Logger, ObjectCount
Defined in:
lib/smith/agent.rb

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from ObjectCount

#object_count

Methods included from Logger

included

Constructor Details

#initialize(uuid) ⇒ Agent

Returns a new instance of Agent.



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
 
# File 'lib/smith/agent.rb', line 11

def initialize(uuid)
  @name = self.class.to_s
  @pid = $$
  @uuid = uuid

  @factory = QueueFactory.new

  @signal_handler = SelfPipe.new(self)

  setup_control_queue

  @start_time = Time.now

  @state = :starting

  @on_stopping = proc {|completion| completion.succeed }
  @on_starting = proc {|completion| completion.succeed }
  @on_running = proc {|completion| completion.succeed }
  @on_exception = proc {}

  @on_starting_completion = EM::Completion.new.tap do |c|
    c.completion do |completion|
      acknowledge_start do
        @on_running.call(@on_running_completion)
        logger.info { "Agent started: #{name}:[#{pid}]." }
      end
    end
  end

  @on_running_completion = EM::Completion.new.tap do |c|
    c.completion do |completion|
      start_keep_alive
      setup_stats_queue
      @state = :running
    end
  end

  @on_stopping_completion = EM::Completion.new.tap do |c|
    c.completion do |completion|
      acknowledge_stop do
        @state = :stopping
        Smith.stop do
          logger.info { "Agent stopped: #{name}:[#{pid}]." }
        end
      end
    end
  end

  EM.threadpool_size = 1

  @on_starting.call(@on_starting_completion)
end

Instance Attribute Details

#name ⇒ Object (readonly)

Returns the value of attribute name.



9
10
11
 
# File 'lib/smith/agent.rb', line 9

def name
  @name
end

#pid ⇒ Object (readonly)

Returns the value of attribute pid.



9
10
11
 
# File 'lib/smith/agent.rb', line 9

def pid
  @pid
end

#uuid ⇒ Object (readonly)

Returns the value of attribute uuid.



9
10
11
 
# File 'lib/smith/agent.rb', line 9

def uuid
  @uuid
end

Class Method Details

.options(opts) ⇒ Object

Options supported: :monitor, the agency will monitor the agent & if dies restart. :singleton, only every have one agent. If this is set to false

multiple agents are allowed.



117
118
119
120
121
 
# File 'lib/smith/agent.rb', line 117

def options(opts)
  opts.each do |k, v|
    Smith.config.agent.send("#{k}=", v)
  end
end

Instance Method Details

#install_signal_handler(signal, position = :end, &blk) ⇒ Object 



91
92
93
 
# File 'lib/smith/agent.rb', line 91

def install_signal_handler(signal, position=:end, &blk)
  @signal_handler.install_signal_handler(signal, position=:end, &blk)
end

#on_exception(&blk) ⇒ Object

The agent may hook into this if they want to do something on exception. It should be noted that, since an exception occured, the reactor will not be running at this point. Even if we restarted the reactor before calling this it would be a different reactor than existed when assigning the block so this would potentially lead to confusion. If the agent really needs the reactor to do something it can always restart the reactor itself.

Parameters:

  • blk (Block) —

    This block will be passed the exception as an argument.



82
83
84
 
# File 'lib/smith/agent.rb', line 82

def on_exception(&blk)
  @on_exception = blk
end

#on_running(&blk) ⇒ Object 



68
69
70
 
# File 'lib/smith/agent.rb', line 68

def on_running(&blk)
  @on_running = blk
end

#on_stopping(&blk) ⇒ Object 



64
65
66
 
# File 'lib/smith/agent.rb', line 64

def on_stopping(&blk)
  @on_stopping = blk
end

#receiver(queue_name, opts = {}, &blk) ⇒ Object 



104
105
106
 
# File 'lib/smith/agent.rb', line 104

def receiver(queue_name, opts={}, &blk)
  queues.receiver(queue_name, opts, &blk)
end

#run ⇒ Object

Override this method to implement your own agent.

Raises:

  • (ArgumentError) 


87
88
89
 
# File 'lib/smith/agent.rb', line 87

def run
  raise ArgumentError, "You must override this method"
end

#sender(queue_names, opts = {}, &blk) ⇒ Object 



108
109
110
 
# File 'lib/smith/agent.rb', line 108

def sender(queue_names, opts={}, &blk)
  Array(queue_names).each { |queue_name| queues.sender(queue_name, opts, &blk) }
end

#state ⇒ Object 



95
96
97
 
# File 'lib/smith/agent.rb', line 95

def state
  @state
end