Class: Pigeon::Engine

Inherits:
Object
  • Object
show all
Extended by:
OptionAccessor
Defined in:
lib/pigeon/engine.rb

Defined Under Namespace

Classes: ConfigurationError, RuntimeError

Constant Summary collapse

CHAINS =

Constants ============================================================

%w[
  after_initialize
  before_start
  after_start
  before_stop
  after_stop
].collect(&:to_sym).freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Methods included from OptionAccessor

option_accessor, option_reader, option_writer

Constructor Details

#initialize(options = nil) ⇒ Engine

Instance Methods =====================================================



240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/pigeon/engine.rb', line 240

def initialize(options = nil)
  @id = Pigeon::Support.unique_id

  @options = options || { }
  
  @task_lock = Mutex.new
  @task_locks = { }

  @task_register_lock = Mutex.new
  @registered_tasks = { }
  
  self.logger ||= self.engine_logger
  self.logger.level = Pigeon::Logger::DEBUG if (self.debug?)
  
  @dispatcher = { }
  
  run_chain(:after_initialize)
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



43
44
45
 
# File 'lib/pigeon/engine.rb', line 43

def id
  @id
end

Class Method Details

.default_engine ⇒ Object

Returns a handle to the engine currently running, or nil if no engine is currently active.



216
217
218
 
# File 'lib/pigeon/engine.rb', line 216

def self.default_engine
  @engines and @engines[0]
end

.engine_logger ⇒ Object

Returns a default logger for the engine.



195
196
197
198
199
200
201
202
 
# File 'lib/pigeon/engine.rb', line 195

def self.engine_logger
  @engine_logger ||= begin
    f = File.open(File.expand_path(self.engine_log_name, self.log_dir), 'a')
    f.sync = true

    Pigeon::Logger.new(f)
  end
end

.execute_in_main_thread(&block) ⇒ Object

Schedules a block for execution on the main EventMachine thread. This is a wrapper around the EventMachine.schedule method.



234
235
236
 
# File 'lib/pigeon/engine.rb', line 234

def self.execute_in_main_thread(&block)
  EventMachine.schedule(&block)
end

.launch(options = nil) ⇒ Object

Launches the engine with the specified options



110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
 
# File 'lib/pigeon/engine.rb', line 110

def self.launch(options = nil)
  engine = nil
  
  EventMachine.run do
    engine = new(options)
    
    yield(engine) if (block_given?)
  
    Signal.trap('INT') do
      engine.terminate
    end

    Pigeon::Engine.register_engine(engine)

    engine.run
  end

  Pigeon::Engine.unregister_engine(engine)
end

.log_dir ⇒ Object

Returns the full path to the directory used to store logs.



105
106
107
 
# File 'lib/pigeon/engine.rb', line 105

def self.log_dir
  @log_file_path ||= Pigeon::Support.find_writable_directory(self.try_log_dirs)
end

.name ⇒ Object

Returns the human-readable name of this engine. Defaults to the name of the engine class, but can be replaced to customize a subclass.



59
60
61
 
# File 'lib/pigeon/engine.rb', line 59

def self.name
  @name or self.to_s.gsub(/::/, ' ')
end

.pid_file ⇒ Object 



130
131
132
 
# File 'lib/pigeon/engine.rb', line 130

def self.pid_file
  @pid_file ||= Pigeon::Pidfile.new(self.pid_file_path)
end

.pid_file_name ⇒ Object

Returns the name of the PID file to use. The full path to the file is specified elsewhere.



88
89
90
 
# File 'lib/pigeon/engine.rb', line 88

def self.pid_file_name
  @pid_file_name or self.name.downcase.gsub(/ /, '-') + '.pid'
end

.pid_file_path ⇒ Object

Returns the full path to the PID file that should be used to track the running status of this engine.



94
95
96
97
98
99
100
101
102
 
# File 'lib/pigeon/engine.rb', line 94

def self.pid_file_path
  @pid_file_path ||= begin
    if (path = Pigeon::Support.find_writable_directory(self.try_pid_dirs))
      File.expand_path(self.pid_file_name, path)
    else
      raise ConfigurationError, "Could not find a writable directory for the PID file in: #{self.try_pid_dirs.join(' ')}"
    end
  end
end

.process_name ⇒ Object

Returns the custom process name for this engine or nil if not assigned.



64
65
66
 
# File 'lib/pigeon/engine.rb', line 64

def self.process_name
  @process_name
end

.process_name=(value) ⇒ Object

Assigns the process name. This will be applied only when the engine is started.



70
71
72
 
# File 'lib/pigeon/engine.rb', line 70

def self.process_name=(value)
  @process_name = value
end

.query_logger ⇒ Object

Returns a default logger for queries.



205
206
207
208
209
210
211
212
 
# File 'lib/pigeon/engine.rb', line 205

def self.query_logger
  @query_logger ||= begin
    f = File.open(File.expand_path(self.query_log_name, self.log_dir), 'a')
    f.sync = true
  
    Pigeon::Logger.new(f)
  end
end

.register_engine(engine) ⇒ Object

Registers the engine as running. The first engine running will show up as the default engine.



222
223
224
225
 
# File 'lib/pigeon/engine.rb', line 222

def self.register_engine(engine)
  @engines ||= [ ]
  @engines << engine
end

.restart ⇒ Object 



177
178
179
180
 
# File 'lib/pigeon/engine.rb', line 177

def self.restart
  self.stop
  self.start
end

.run {|$$| ... } ⇒ Object

Yields:

  • ($$) 


150
151
152
153
154
 
# File 'lib/pigeon/engine.rb', line 150

def self.run
  yield($$) if (block_given?)

  launch(:foreground => true)
end

.run_chain(chain_name, instance) ⇒ Object 



373
374
375
376
377
378
379
380
381
 
# File 'lib/pigeon/engine.rb', line 373

def run_chain(chain_name, instance)
  chain = instance_variable_get(:"@_#{chain_name}_chain")

  return unless (chain)

  chain.each do |proc|
    instance.instance_eval(&proc)
  end
end

.running? ⇒ Boolean

Returns:

  • (Boolean) 


182
183
184
 
# File 'lib/pigeon/engine.rb', line 182

def self.running?
  pid_file.running
end

.start(options = nil) {|pid| ... } ⇒ Object

Yields:

  • (pid) 


134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
 
# File 'lib/pigeon/engine.rb', line 134

def self.start(options = nil)
  logger = self.engine_logger
  
  pid = Pigeon::Support.daemonize(logger) do
    launch({
      :logger => logger
    }.merge(options || { }))
  end

  pid_file.create!(pid)

  yield(pid) if (block_given?)
  
  pid
end

.status {|pid| ... } ⇒ Object

Yields:

  • (pid) 


186
187
188
189
190
191
192
 
# File 'lib/pigeon/engine.rb', line 186

def self.status
  pid = pid_file.running
  
  yield(pid) if (block_given?)
  
  pid
end

.stop {|pid| ... } ⇒ Object

Yields:

  • (pid) 


156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/pigeon/engine.rb', line 156

def self.stop
  pid = self.pid_file.running
  
  if (pid)
    begin
      Process.kill('INT', pid)
    rescue Errno::ESRCH
      # No such process exception
      pid = nil
    end

    pid_file.remove!
  end
  
  pid = pid.to_i if (pid)

  yield(pid) if (block_given?)
  
  pid
end

.unregister_engine(engine) ⇒ Object

Removes the engine from the list of running engines.



228
229
230
 
# File 'lib/pigeon/engine.rb', line 228

def self.unregister_engine(engine)
  @engines.delete(engine)
end

.user ⇒ Object

Returns the user this process should run as, or nil if no particular user is required. This will be applied after the engine has been started and the after_start call has been triggered.



77
78
79
 
# File 'lib/pigeon/engine.rb', line 77

def self.user
  @user
end

.user=(value) ⇒ Object

Assigns the user this process should run as, given a username.



82
83
84
 
# File 'lib/pigeon/engine.rb', line 82

def self.user=(value)
  @user = value
end

Instance Method Details

#debug? ⇒ Boolean

Returns true if the debug option was set, false otherwise.

Returns:

  • (Boolean) 


385
386
387
 
# File 'lib/pigeon/engine.rb', line 385

def debug?
  !!self.debug
end

#defer(&block) ⇒ Object

Used to defer a block of work for near-immediate execution. Is a wrapper around EventMachine.defer and does not perform as well as using the alternate dispatch method.



326
327
328
 
# File 'lib/pigeon/engine.rb', line 326

def defer(&block)
  EventMachine.defer(&block)
end

#dispatch(name = :default, &block) ⇒ Object

Used to dispatch a block for immediate processing on a background thread. An optional queue name can be used to sequence tasks properly. The main queue has a large number of threads, while the named queues default to only one so they can be processed sequentially.



350
351
352
353
354
 
# File 'lib/pigeon/engine.rb', line 350

def dispatch(name = :default, &block)
  target_queue = @dispatcher[name] ||= Pigeon::Dispatcher.new(name == :default ? nil : 1)
  
  target_queue.perform(&block)
end

#execute_in_main_thread(&block) ⇒ Object

Schedules a block for execution on the main EventMachine thread. This is a wrapper around the EventMachine.schedule method.



332
333
334
 
# File 'lib/pigeon/engine.rb', line 332

def execute_in_main_thread(&block)
  EventMachine.schedule(&block)
end

#foreground? ⇒ Boolean

Returns true if running in the foreground, false otherwise.

Returns:

  • (Boolean) 


390
391
392
 
# File 'lib/pigeon/engine.rb', line 390

def foreground?
  !!self.foreground
end

#host ⇒ Object

Returns the hostname of the system this engine is running on.



260
261
262
 
# File 'lib/pigeon/engine.rb', line 260

def host
  Socket.gethostname
end

#periodically(interval, &block) ⇒ Object

Periodically calls a block. No check is performed to see if the block is already executing.



319
320
321
 
# File 'lib/pigeon/engine.rb', line 319

def periodically(interval, &block)
  EventMachine::PeriodicTimer.new(interval, &block)
end

#periodically_trigger_task(task_name = nil, interval = 1, &block) ⇒ Object

Used to periodically execute a task or block. When giving a task name, a method by that name is called, otherwise a block must be supplied. An interval can be specified in seconds, or will default to 1.



283
284
285
286
287
 
# File 'lib/pigeon/engine.rb', line 283

def periodically_trigger_task(task_name = nil, interval = 1, &block)
  periodically(interval) do
    trigger_task(task_name, &block)
  end
end

#register_task(task) ⇒ Object

Registers a task with the engine. The given task will then be included in the list returned by registered_tasks.



396
397
398
399
400
 
# File 'lib/pigeon/engine.rb', line 396

def register_task(task)
  @task_register_lock.synchronize do
    @registered_tasks[task] = task
  end
end

#registered_tasks ⇒ Object

Returns a list of tasks that have been registered with the engine.



410
411
412
413
414
 
# File 'lib/pigeon/engine.rb', line 410

def registered_tasks
  @task_register_lock.synchronize do
    @registered_tasks.values
  end
end

#run ⇒ Object

Handles the run phase of the engine, triggers the before_start and after_start events accordingly.



266
267
268
269
270
271
272
273
274
275
276
277
278
 
# File 'lib/pigeon/engine.rb', line 266

def run
  assign_process_name!

  run_chain(:before_start)

  STDOUT.sync = true

  logger.info("Engine \##{id} Running")
  
  run_chain(:after_start)
  
  switch_to_effective_user! if (self.class.user)
end

#task_lock(task_name) ⇒ Object

This is a somewhat naive locking mechanism that may break down when two requests are fired off within a nearly identical period. For now, this achieves a general purpose solution that should work under most circumstances. Refactor later to improve.



301
302
303
304
305
306
307
308
309
310
311
 
# File 'lib/pigeon/engine.rb', line 301

def task_lock(task_name)
  @task_lock.synchronize do
    @task_locks[task_name] ||= Mutex.new
  end
  
  return if (@task_locks[task_name].locked?)
  
  @task_locks[task_name].synchronize do
    yield if (block_given?)
  end
end

#terminate ⇒ Object

Shuts down the engine. Will also trigger the before_stop and after_stop events.



338
339
340
341
342
343
344
 
# File 'lib/pigeon/engine.rb', line 338

def terminate
  run_chain(:before_stop)

  EventMachine.stop_event_loop

  run_chain(:after_stop)
end

#timer(interval, &block) ⇒ Object 



313
314
315
 
# File 'lib/pigeon/engine.rb', line 313

def timer(interval, &block)
  EventMachine::Timer.new(interval, &block)
end

#trigger_task(task_name = nil, &block) ⇒ Object

This acts as a lock to prevent over-lapping calls to the same method. While the first call is in progress, all subsequent calls will be ignored.



291
292
293
294
295
 
# File 'lib/pigeon/engine.rb', line 291

def trigger_task(task_name = nil, &block)
  task_lock(task_name || block) do
    block_given? ? yield : send(task_name)
  end
end

#unregister_task(task) ⇒ Object

Removes a task from the list of tasks registered with this engine.



403
404
405
406
407
 
# File 'lib/pigeon/engine.rb', line 403

def unregister_task(task)
  @task_register_lock.synchronize do
    @registered_tasks.delete(task)
  end
end