Class: Nl::BlockingTransport

Inherits:
Object
  • Object
show all
Defined in:
nl/lib/nl/blocking_transport.rb

Overview

Drives one exchange at a time with blocking socket operations.

Defined Under Namespace

Classes: ConcurrentOperationError, UnexpectedSequenceError

Instance Method Summary collapse

Constructor Details

#initialize(socket, protocol:, notifications:) ⇒ BlockingTransport

Returns a new instance of BlockingTransport.



21
22
23
24
25
26
27
 
# File 'nl/lib/nl/blocking_transport.rb', line 21

def initialize(socket, protocol:, notifications:)
  @socket = socket
  @protocol = protocol
  @sequences = SequenceAllocator.new
  @mutex = Mutex.new
  @notifications = notifications
end

Instance Method Details

#async_capable? ⇒ Boolean

: false

Returns:

  • (Boolean) 


61
62
63
 
# File 'nl/lib/nl/blocking_transport.rb', line 61

def async_capable? #: false
  false
end

#close ⇒ Object

: void



93
94
95
96
 
# File 'nl/lib/nl/blocking_transport.rb', line 93

def close #: void
  @socket.close unless @socket.closed?
  nil
end

#exchange(endpoint, kind, request_class, reply_class, args) ⇒ Object 



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
 
# File 'nl/lib/nl/blocking_transport.rb', line 29

def exchange(endpoint, kind, request_class, reply_class, args)
  unless locked = @mutex.try_lock
    raise ConcurrentOperationError, 'BlockingTransport supports only one active operation'
  end

  request = @protocol.build_request(endpoint, kind, request_class, args)
  seq = @sequences.next
  pid = @socket.local_port_id
  key = [seq, pid]
  @protocol.send_message(@socket, endpoint, request, seq:, pid:)
  mode = kind == :dump ? :dump : (reply_class ? :reply : :no_reply)
  exchange = Exchange.new(mode:)
  result = [] unless block_given?

  until exchange.complete?
    receive(endpoint, key, reply_class) do |message|
      case outcome = exchange.accept(message)
      when Exchange::Item
        block_given? ? yield(outcome.value) : result << outcome.value
      when Exchange::Failure
        raise outcome.exception
      end
    end
  end

  return if block_given?

  kind == :dump ? result : exchange.result
ensure
  @mutex.unlock if locked
end

#receive_notification(endpoint, timeout: nil) ⇒ Object 



65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'nl/lib/nl/blocking_transport.rb', line 65

def receive_notification(endpoint, timeout: nil)
  channel = @notifications.channel(endpoint)
  return channel.pop(timeout: 0)
rescue TimeoutError
  unless locked = @mutex.try_lock
    raise ConcurrentOperationError, 'BlockingTransport supports only one active operation'
  end

  deadline = Process.clock_gettime(Process::CLOCK_MONOTONIC) + timeout if timeout
  loop do
    begin
      return channel.pop(timeout: 0)
    rescue TimeoutError
      # Read another datagram below.
    end

    remaining = deadline && deadline - Process.clock_gettime(Process::CLOCK_MONOTONIC)
    raise TimeoutError, 'notification receive timed out' if remaining && remaining <= 0
    if remaining && !@socket.wait_readable(remaining)
      raise TimeoutError, 'notification receive timed out'
    end

    receive_notifications
  end
ensure
  @mutex.unlock if locked
end