Class: OMQ::Ractor::Context

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/ractor.rb

Overview

Frozen, shareable context passed to the worker Ractor. The user calls #sockets to trigger the setup handshake.

An optional data object (any Ractor.make_shareable-able value) can be passed via OMQ::Ractor.new(data: …) and retrieved inside the worker block with omq.data. This is the only way to pass extra information into a worker block under Ruby 4.0's strict Ractor isolation, which forbids capturing any outer local variable.

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(setup_port, output_ports, socket_configs, data: nil) ⇒ Context

Returns a new instance of Context.

Parameters:

  • setup_port (Ractor::Port) —

    port for the setup handshake

  • output_ports (Array<Ractor::Port, nil>) —

    output ports for writable sockets

  • socket_configs (Array<Hash>) —

    per-socket configuration hashes

  • data (Object, nil) (defaults to: nil) —

    optional shareable data for the worker



341
342
343
344
345
346
347
 
# File 'lib/omq/ractor.rb', line 341

def initialize(setup_port, output_ports, socket_configs, data: nil)
  @setup_port     = setup_port
  @output_ports   = output_ports
  @socket_configs = socket_configs
  @data           = data
  ::Ractor.make_shareable(self)
end

Instance Attribute Details

#data ⇒ Object? (readonly)

User-supplied shareable data (passed as data: to OMQ::Ractor.new).

Returns:

  • (Object, nil) 


354
355
356
 
# File 'lib/omq/ractor.rb', line 354

def data
  @data
end

Instance Method Details

#sockets ⇒ Array<SocketProxy>

Performs the setup handshake and returns SocketProxy objects.

Returns:



360
361
362
363
364
365
366
367
368
369
 
# File 'lib/omq/ractor.rb', line 360

def sockets
  input_ports = @socket_configs.map { |cfg| cfg[:readable] ? ::Ractor::Port.new : nil }

  @setup_port.send(input_ports)

  proxies = @socket_configs.each_with_index.map do |cfg, i|
    SocketProxy.new(input_ports[i], @output_ports[i], cfg[:topic_type])
  end
  SocketSet.new(proxies)
end