Module: OMQ::Transport::Inproc

Defined in:
lib/omq/transport/inproc.rb,
lib/omq/transport/inproc/pipe.rb

Overview

Ruby in-process transport (ruby:// scheme).

Both peers are Ruby backend sockets in the same process. Messages are transferred as Ruby arrays — no ZMTP framing, no byte serialization. Parts are already frozen by Writable#send, so the receiver sees the same immutable contract as ZMTP transports.

The inproc:// scheme is reserved for native backends (FFI/libzmq, Rust/omq-tokio) which have their own in-process registries.

Defined Under Namespace

Classes: Listener, Pipe

Constant Summary collapse

COMMAND_TYPES =

Socket types that exchange commands (SUBSCRIBE/CANCEL) over inproc.

%i[PUB SUB XPUB XSUB RADIO DISH].freeze

Class Attribute Summary collapse

Class Method Summary collapse

Class Attribute Details

.registry ⇒ Hash{String => Engine} (readonly)

Returns bound inproc endpoints.

Returns:

  • (Hash{String => Engine}) —

    bound inproc endpoints



38
39
40
 
# File 'lib/omq/transport/inproc.rb', line 38

def registry
  @registry
end

Class Method Details

.connect(endpoint, engine) ⇒ void

This method returns an undefined value.

Connects to a bound inproc endpoint.

Parameters:

  • endpoint (String) —

    e.g. "ruby://my-endpoint"

  • engine (Engine) —

    the connecting engine



70
71
72
73
74
 
# File 'lib/omq/transport/inproc.rb', line 70

def connect(endpoint, engine, **)
  bound_engine = @mutex.synchronize { @registry[endpoint] }
  bound_engine ||= await_bind(endpoint, engine) or return
  establish_link(engine, bound_engine, endpoint)
end

.listener(endpoint, engine) ⇒ Listener

Creates a bound inproc listener.

Parameters:

  • endpoint (String) —

    e.g. "ruby://my-endpoint"

  • engine (Engine) —

    the owning engine

Returns:

Raises:

  • (ArgumentError) —

    if endpoint is already bound



48
49
50
51
52
53
54
55
56
57
58
59
60
61
 
# File 'lib/omq/transport/inproc.rb', line 48

def listener(endpoint, engine, **)
  @mutex.synchronize do
    if @registry.key?(endpoint)
      raise ArgumentError, "endpoint already bound: #{endpoint}"
    end

    @registry[endpoint] = engine

    # Wake any pending connects
    @waiters[endpoint].each { |p| p.resolve(true) }
    @waiters.delete(endpoint)
  end
  Listener.new(endpoint)
end

.reset! ⇒ void

This method returns an undefined value.

Resets the registry. Used in tests.



91
92
93
94
95
96
 
# File 'lib/omq/transport/inproc.rb', line 91

def reset!
  @mutex.synchronize do
    @registry.clear
    @waiters.clear
  end
end

.unbind(endpoint) ⇒ void

This method returns an undefined value.

Removes a bound endpoint from the registry.

Parameters:

  • endpoint (String) 


82
83
84
 
# File 'lib/omq/transport/inproc.rb', line 82

def unbind(endpoint)
  @mutex.synchronize { @registry.delete(endpoint) }
end