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
  before_resume
  after_resume
  before_standby
  after_standby
  before_shutdown
  after_shutdown
].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 =====================================================



260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
 
# File 'lib/pigeon/engine.rb', line 260

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

  wrap_chain(:initialize) do
    @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 = { }
  
    @state = :initialized
  end
end

Instance Attribute Details

#id ⇒ Object (readonly)

Returns the value of attribute id.



48
49
50
 
# File 'lib/pigeon/engine.rb', line 48

def id
  @id
end

#state ⇒ Object (readonly)

Returns the value of attribute state.



49
50
51
 
# File 'lib/pigeon/engine.rb', line 49

def state
  @state
end

Class Method Details

.chain_procs(chain_name) ⇒ Object 



435
436
437
 
# File 'lib/pigeon/engine.rb', line 435

def chain_procs(chain_name)
  instance_variable_get(:"@_#{chain_name}_chain")
end

.default_engine ⇒ Object

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



236
237
238
 
# File 'lib/pigeon/engine.rb', line 236

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

.engine_logger ⇒ Object

Returns a default logger for the engine.



215
216
217
218
219
220
221
222
 
# File 'lib/pigeon/engine.rb', line 215

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, self.log_rotation)
  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.



254
255
256
 
# File 'lib/pigeon/engine.rb', line 254

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

.launch(options = nil) ⇒ Object

Launches the engine with the specified options



122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/pigeon/engine.rb', line 122

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.



117
118
119
 
# File 'lib/pigeon/engine.rb', line 117

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.



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

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

.pid_file ⇒ Object 



142
143
144
 
# File 'lib/pigeon/engine.rb', line 142

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.



100
101
102
 
# File 'lib/pigeon/engine.rb', line 100

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.



106
107
108
109
110
111
112
113
114
 
# File 'lib/pigeon/engine.rb', line 106

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.



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

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.



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

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

.query_logger ⇒ Object

Returns a default logger for queries.



225
226
227
228
229
230
231
232
 
# File 'lib/pigeon/engine.rb', line 225

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, self.log_rotation)
  end
end

.register_engine(engine) ⇒ Object

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



242
243
244
245
 
# File 'lib/pigeon/engine.rb', line 242

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

.restart ⇒ Object 



197
198
199
200
 
# File 'lib/pigeon/engine.rb', line 197

def self.restart
  self.stop
  self.start
end

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

Yields:

  • ($$) 


162
163
164
165
166
 
# File 'lib/pigeon/engine.rb', line 162

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

  launch(:foreground => true)
end

.running? ⇒ Boolean

Returns:

  • (Boolean) 


202
203
204
 
# File 'lib/pigeon/engine.rb', line 202

def self.running?
  pid_file.running
end

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

Yields:

  • (pid) 


146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
 
# File 'lib/pigeon/engine.rb', line 146

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) 


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

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

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

Yields:

  • (pid) 


168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
 
# File 'lib/pigeon/engine.rb', line 168

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
    
    begin
      while (Process.kill(0, pid))
        sleep(1)
      end
    rescue Errno::ESRCH
      # No such process, already terminated
    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.



248
249
250
 
# File 'lib/pigeon/engine.rb', line 248

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.



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

def self.user
  @user
end

.user=(value) ⇒ Object

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



94
95
96
 
# File 'lib/pigeon/engine.rb', line 94

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

Instance Method Details

#debug? ⇒ Boolean

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

Returns:

  • (Boolean) 


441
442
443
 
# File 'lib/pigeon/engine.rb', line 441

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.



348
349
350
 
# File 'lib/pigeon/engine.rb', line 348

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.



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

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

    target_queue.perform(&block)
  else
    EventMachine.next_tick(&block)
  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.



354
355
356
 
# File 'lib/pigeon/engine.rb', line 354

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

#foreground? ⇒ Boolean

Returns true if running in the foreground, false otherwise.

Returns:

  • (Boolean) 


446
447
448
 
# File 'lib/pigeon/engine.rb', line 446

def foreground?
  !!self.foreground
end

#host ⇒ Object

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



282
283
284
 
# File 'lib/pigeon/engine.rb', line 282

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.



341
342
343
 
# File 'lib/pigeon/engine.rb', line 341

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.



305
306
307
308
309
 
# File 'lib/pigeon/engine.rb', line 305

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.



452
453
454
455
456
 
# File 'lib/pigeon/engine.rb', line 452

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.



466
467
468
469
470
 
# File 'lib/pigeon/engine.rb', line 466

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

#resume! ⇒ Object 



381
382
383
384
385
386
387
388
389
390
391
392
 
# File 'lib/pigeon/engine.rb', line 381

def resume!
  case (@state)
  when :running
    # Ignored since already running.
  when :terminated
    # Invalid operation, should produce error.
  else
    wrap_chain(:resume) do
      @state = :running
    end
  end
end

#run ⇒ Object

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



288
289
290
291
292
293
294
295
296
297
298
299
300
 
# File 'lib/pigeon/engine.rb', line 288

def run
  assign_process_name!

  wrap_chain(:start) do
    STDOUT.sync = true

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

    @state = :running
  end
end

#shutdown! ⇒ Object 



407
408
409
410
411
412
413
414
415
416
 
# File 'lib/pigeon/engine.rb', line 407

def shutdown!
  case (@state)
  when :terminated
    # Already terminated, ignored.
  else
    wrap_chain(:shutdown) do
      self.terminate
    end
  end
end

#standby! ⇒ Object 



394
395
396
397
398
399
400
401
402
403
404
405
 
# File 'lib/pigeon/engine.rb', line 394

def standby!
  case (@state)
  when :standby
    # Already in standby state, ignored.
  when :terminated
    # Invalid operation, should produce error.
  else
    wrap_chain(:standby) do
      @state = :standby
    end
  end
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.



323
324
325
326
327
328
329
330
331
332
333
 
# File 'lib/pigeon/engine.rb', line 323

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.



360
361
362
363
364
365
 
# File 'lib/pigeon/engine.rb', line 360

def terminate
  wrap_chain(:stop) do
    EventMachine.stop_event_loop
    @state = :terminated
  end
end

#timer(interval, &block) ⇒ Object 



335
336
337
 
# File 'lib/pigeon/engine.rb', line 335

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.



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

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.



459
460
461
462
463
 
# File 'lib/pigeon/engine.rb', line 459

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