Class: Nl::Async::Dispatcher

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

Overview

Owns the single receive loop for a socket and routes replies by sequence.

Instance Method Summary collapse

Constructor Details

#initialize(socket, protocol:, executor: :thread, notifications:) ⇒ Dispatcher

Returns a new instance of Dispatcher.



14
15
16
17
18
19
20
21
22
23
24
25
 
# File 'nl/lib/nl/async/dispatcher.rb', line 14

def initialize(socket, protocol:, executor: :thread, notifications:)
  @socket = socket
  @protocol = protocol
  @sequences = SequenceAllocator.new
  @notifications = notifications
  @driver = executor.respond_to?(:start) ? executor : Async.driver(executor)
  @mutex = Mutex.new
  @send_mutex = Mutex.new
  @pending = {}
  @closed = false
  @task = @driver.start { receive_loop }
end

Instance Method Details

#async_capable? ⇒ Boolean

: true

Returns:

  • (Boolean) 


75
76
77
 
# File 'nl/lib/nl/async/dispatcher.rb', line 75

def async_capable? #: true
  true
end

#close ⇒ Object 



83
84
85
86
87
88
89
90
91
92
93
94
 
# File 'nl/lib/nl/async/dispatcher.rb', line 83

def close
  task = @mutex.synchronize do
    next if @closed

    @closed = true
    @socket.close unless @socket.closed?
    @task
  end
  fail_all(ClosedError.new('dispatcher is closed'))
  @driver.stop(task) if task && !task.equal?(::Thread.current)
  nil
end

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



27
28
29
30
31
32
33
34
35
36
37
38
39
 
# File 'nl/lib/nl/async/dispatcher.rb', line 27

def exchange(endpoint, kind, request_class, reply_class, args, &block)
  operation = exchange_async(endpoint, kind, request_class, reply_class, args)
  if kind == :dump
    if block
      operation.each(&block)
      nil
    else
      operation.to_a
    end
  else
    operation.await
  end
end

#exchange_async(endpoint, kind, request_class, reply_class, args, stream_capacity: nil) ⇒ Object 



41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
 
# File 'nl/lib/nl/async/dispatcher.rb', line 41

def exchange_async(endpoint, kind, request_class, reply_class, args, stream_capacity: nil)
  request = @protocol.build_request(endpoint, kind, request_class, args)

  key = nil
  seq = pid = nil
  operation = sink = nil
  @send_mutex.synchronize do
    @mutex.synchronize do
      raise ClosedError, 'dispatcher is closed' if @closed

      pid = @socket.local_port_id
      seq = @sequences.next { @pending.key?([it, pid]) }
      key = [seq, pid]
      mailbox = Mailbox.new(capacity: stream_capacity)
      operation, sink = if kind == :dump
        Stream.build(mailbox:, on_close: -> { discard(key) })
      else
        Future.build(mailbox:, on_close: -> { discard(key) })
      end
      mode = kind == :dump ? :dump : (reply_class ? :reply : :no_reply)
      exchange = Exchange.new(mode:)
      @pending[key] = Pending.new(exchange:, endpoint:, reply_class:, sink:)
    end

    begin
      @protocol.send_message(@socket, endpoint, request, seq:, pid:)
    rescue Exception => error
      fail_pending(key, error)
      raise
    end
  end
  operation
end

#receive_notification(endpoint, timeout: nil) ⇒ Object 



79
80
81
 
# File 'nl/lib/nl/async/dispatcher.rb', line 79

def receive_notification(endpoint, timeout: nil)
  @notifications.channel(endpoint).pop(timeout:)
end