Class: Rutema::Dispatcher

Inherits:
Object
  • Object
show all
Defined in:
lib/rutema/core/engine.rb

Overview

rubocop:enable Naming/PredicatePrefix

Class functioning as a de-multiplexer between the Engine and the various Reporters instances

In stream mode the incoming queue is popped periodically and the messages are distributed to the queues of any subscribed event reporters. By default this includes Reporters::Collector which is then used at the end of a run to provide the collected data to all registered block mode reporters

Constant Summary collapse

INTERVAL =

The interval between queue operations

0.01

Instance Method Summary collapse

Constructor Details

#initialize(queue, configuration) ⇒ Dispatcher

Initialize a new demultiplexer and instantiate all reporters requested by the passed configuration

  • queue - the queue which will be shared between the Engine instance and the Reporter instances
  • configuration - the Configuration instance of the rutema run


248
249
250
251
252
253
254
255
256
257
258
259
260
261
 
# File 'lib/rutema/core/engine.rb', line 248

def initialize(queue, configuration)
  @queue = queue
  @queues = {}
  @streaming_reporters = []
  @block_reporters = []
  @collector = Rutema::Reporters::Collector.new(nil, self)
  if configuration.reporters
    instances = configuration.reporters.values.map { |v| instantiate_reporter(v, configuration) if v[:class] != Reporters::Summary }.compact
    @streaming_reporters, = instances.partition { |rep| rep.respond_to?(:update) }
    @block_reporters, = instances.partition { |rep| rep.respond_to?(:report) }
  end
  @streaming_reporters << @collector
  @configuration = configuration
end

Instance Method Details

#exit ⇒ Object

Dispatch all messages in the incoming queue to the subscribed reporters, exit all streaming reporters' threads and then stop the own internal dispatch thread



305
306
307
308
309
310
311
312
 
# File 'lib/rutema/core/engine.rb', line 305

def exit
  puts "Exiting main dispatcher" if $DEBUG
  return unless @thread

  flush
  @streaming_reporters.each(&:exit)
  Thread.kill(@thread)
end

#report(specs) ⇒ Object

Call all block reporters' BlockReporter#report method



294
295
296
297
298
299
 
# File 'lib/rutema/core/engine.rb', line 294

def report(specs)
  @block_reporters.each do |r|
    r.report(specs, @collector.states, @collector.errors)
  end
  Reporters::Summary.new(@configuration, self).report(specs, @collector.states, @collector.errors)
end

#run! ⇒ Object

Start #update threads of all event/streaming reporters and then start a new locally managed thread which continually dispatches messages from the incoming queue



281
282
283
284
285
286
287
288
289
290
 
# File 'lib/rutema/core/engine.rb', line 281

def run!
  puts "Running #{@streaming_reporters.size} streaming reporters" if $DEBUG
  @streaming_reporters.each(&:run!)
  @thread = Thread.new do
    loop do
      dispatch
      sleep INTERVAL
    end
  end
end

#subscribe(identifier) ⇒ Object

Call this to establish a queue with the given identifier

This method will create a new queue within the Dispatcher instance into which data from the incoming queue from the Engine instance will be dispatched to.

  • identifier - a unique identifier for the queue. If two identifiers collide the new subscriber will replace the earlier one


272
273
274
275
 
# File 'lib/rutema/core/engine.rb', line 272

def subscribe(identifier)
  @queues[identifier] = Queue.new
  return @queues[identifier]
end