Class: NNQ::CLI::BaseRunner

Inherits:
Object
  • Object
show all
Defined in:
lib/nnq/cli/base_runner.rb

Overview

Template runner base class for all socket-type CLI runners. Subclasses override #run_loop to implement socket-specific behaviour.

nnq carries one String body per message (no multipart). The runner protocol wraps it in a 1-element Array internally so that eval expressions using ‘$F`/`$_` work the same way as in omq-cli, and unwraps to a bare String at the send boundary.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, socket_class) ⇒ BaseRunner

Returns a new instance of BaseRunner.

Parameters:

  • config (Config) —

    frozen CLI configuration

  • socket_class (Class) —

    NNQ socket class to instantiate (e.g. NNQ::PUSH)



20
21
22
23
24
 
# File 'lib/nnq/cli/base_runner.rb', line 20

def initialize(config, socket_class)
  @config = config
  @klass  = socket_class
  @fmt    = Formatter.new(config.format, compress: config.compress)
end

Instance Attribute Details

#config ⇒ Config, Object (readonly)

Returns:

  • (Config) —

    frozen CLI configuration

  • (Object) —

    the NNQ socket instance



15
16
17
 
# File 'lib/nnq/cli/base_runner.rb', line 15

def config
  @config
end

#sock ⇒ Config, Object (readonly)

Returns:

  • (Config) —

    frozen CLI configuration

  • (Object) —

    the NNQ socket instance



15
16
17
 
# File 'lib/nnq/cli/base_runner.rb', line 15

def sock
  @sock
end

Instance Method Details

#call(task) ⇒ void

This method returns an undefined value.

Runs the full lifecycle: socket setup, peer wait, BEGIN/END blocks, and the main loop.

Parameters:

  • task (Async::Task) —

    the parent async task



31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/nnq/cli/base_runner.rb', line 31

def call(task)
  set_process_title
  setup_socket
  start_event_monitor if config.verbose >= 2
  maybe_start_transient_monitor(task)
  sleep(config.delay) if config.delay && config.recv_only?
  wait_for_peer if needs_peer_wait?
  run_begin_blocks
  run_loop(task)
  run_end_blocks
ensure
  @sock&.close
end