Class: OMQ::Routing::Req

Inherits:
Object
  • Object
show all
Includes:
RoundRobin
Defined in:
lib/omq/routing/req.rb

Overview

REQ socket routing: round-robin send with strict send/recv alternation.

REQ prepends an empty delimiter frame on send and strips it on receive.

Constant Summary collapse

EMPTY_BINARY =

Shared frozen empty binary string to avoid repeated allocations.

::Protocol::ZMTP::Codec::EMPTY_BINARY

Constants included from RoundRobin

OMQ::Routing::RoundRobin::BATCH_BYTE_CAP, OMQ::Routing::RoundRobin::BATCH_MSG_CAP

Instance Attribute Summary collapse

Instance Method Summary collapse

Methods included from RoundRobin

#send_queues_drained?

Constructor Details

#initialize(engine) ⇒ Req

Returns a new instance of Req.

Parameters:



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

def initialize(engine)
  @engine     = engine
  @recv_queue = Routing.build_queue(engine.options.recv_hwm, :block)
  @state      = :ready        # :ready or :waiting_reply
  init_round_robin(engine)
end

Instance Attribute Details

#recv_queue ⇒ Async::LimitedQueue (readonly)

Returns:

  • (Async::LimitedQueue) 


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

def recv_queue
  @recv_queue
end

Instance Method Details

#connection_added(connection) ⇒ Object

Parameters:

  • connection (Connection) 


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

def connection_added(connection)
  @engine.start_recv_pump(connection, @recv_queue) do |msg|
    @state = :ready
    msg.first&.empty? ? msg[1..] : msg
  end

  add_round_robin_send_connection(connection)
end

#connection_removed(connection) ⇒ Object

Parameters:

  • connection (Connection) 


63
64
65
66
 
# File 'lib/omq/routing/req.rb', line 63

def connection_removed(connection)
  @connections.delete(connection)
  remove_round_robin_send_connection(connection)
end

#dequeue_recv ⇒ Array<String>?

Dequeues the next received message. Blocks until one is available.

Returns:

  • (Array<String>, nil) 


35
36
37
 
# File 'lib/omq/routing/req.rb', line 35

def dequeue_recv
  @recv_queue.dequeue
end

#enqueue(parts) ⇒ Object

Parameters:

  • parts (Array<String>)

Raises:

  • (SocketError) 


71
72
73
74
75
 
# File 'lib/omq/routing/req.rb', line 71

def enqueue(parts)
  raise SocketError, "REQ socket expects send/recv/send/recv order" unless @state == :ready
  @state = :waiting_reply
  enqueue_round_robin(parts)
end

#unblock_recv ⇒ void

This method returns an undefined value.

Wakes a blocked #dequeue_recv with a nil sentinel.



44
45
46
 
# File 'lib/omq/routing/req.rb', line 44

def unblock_recv
  @recv_queue.enqueue(nil)
end