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.

inproc:// is treated as a compatibility alias by the Ruby backend unless a backend or plugin registers a real inproc transport.

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

.canonical_endpoint(endpoint) ⇒ String

Canonicalizes compatibility aliases so ruby:// and inproc:// share the same in-process endpoint namespace.

Parameters:

  • endpoint (String)

Returns:

  • (String) 


47
48
49
 
# File 'lib/omq/transport/inproc.rb', line 47

def canonical_endpoint(endpoint)
  endpoint.sub(/\Ainproc:\/\//, "ruby://")
end

.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



83
84
85
86
87
88
 
# File 'lib/omq/transport/inproc.rb', line 83

def connect(endpoint, engine, **)
  endpoint     = canonical_endpoint(endpoint)
  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



59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
 
# File 'lib/omq/transport/inproc.rb', line 59

def listener(endpoint, engine, **)
  endpoint = canonical_endpoint(endpoint)

  @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.



106
107
108
109
110
111
 
# File 'lib/omq/transport/inproc.rb', line 106

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) 


96
97
98
99
 
# File 'lib/omq/transport/inproc.rb', line 96

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