Class: OMQ::Routing::Pair

Inherits:
Object
  • Object
show all
Includes:
FairRecv
Defined in:
lib/omq/routing/pair.rb

Overview

PAIR socket routing: exclusive 1-to-1 bidirectional.

Only one peer connection is allowed at a time. The send queue is socket-level (one shared bounded queue), and a single send pump fiber drains it into the connected peer. On disconnect, the in-flight batch is dropped (matching libzmq).

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from FairRecv

#dequeue_recv, #unblock_recv

Constructor Details

#initialize(engine) ⇒ Pair

Returns a new instance of Pair.

Parameters:



23
24
25
26
27
28
29
30
 
# File 'lib/omq/routing/pair.rb', line 23

def initialize(engine)
  @engine     = engine
  @connection = nil
  @recv_queue = FairQueue.new
  @send_queue = Routing.build_queue(@engine.options.send_hwm, :block)
  @send_pump  = nil
  @tasks      = []
end

Instance Attribute Details

#recv_queue ⇒ FairQueue (readonly)

Returns:



18
19
20
 
# File 'lib/omq/routing/pair.rb', line 18

def recv_queue
  @recv_queue
end

Instance Method Details

#connection_added(connection) ⇒ Object

Parameters:

  • connection (Connection)

Raises:

  • (RuntimeError) —

    if a connection already exists



36
37
38
39
40
41
42
43
44
45
 
# File 'lib/omq/routing/pair.rb', line 36

def connection_added(connection)
  raise "PAIR allows only one peer" if @connection
  @connection = connection

  add_fair_recv_connection(connection)

  unless connection.is_a?(Transport::Inproc::DirectPipe)
    start_send_pump(connection)
  end
end

#connection_removed(connection) ⇒ Object

Parameters:

  • connection (Connection) 


50
51
52
53
54
55
56
57
 
# File 'lib/omq/routing/pair.rb', line 50

def connection_removed(connection)
  if @connection == connection
    @connection = nil
    @recv_queue.remove_queue(connection)
    @send_pump&.stop
    @send_pump = nil
  end
end

#enqueue(parts) ⇒ Object

Parameters:

  • parts (Array<String>) 


62
63
64
65
66
67
68
69
 
# File 'lib/omq/routing/pair.rb', line 62

def enqueue(parts)
  conn = @connection
  if conn.is_a?(Transport::Inproc::DirectPipe) && conn.direct_recv_queue
    conn.send_message(parts)
  else
    @send_queue.enqueue(parts)
  end
end

#send_queues_drained? ⇒ Boolean

Returns true when the shared send queue is empty.

Returns:

  • (Boolean) —

    true when the shared send queue is empty



84
85
86
 
# File 'lib/omq/routing/pair.rb', line 84

def send_queues_drained?
  @send_queue.empty?
end

#stop ⇒ void

This method returns an undefined value.

Stops all background tasks.



76
77
78
79
 
# File 'lib/omq/routing/pair.rb', line 76

def stop
  @tasks.each(&:stop)
  @tasks.clear
end