Class: OMQ::Rust::Engine

Inherits:
Object
  • Object
show all
Defined in:
lib/omq/rust/engine.rb

Defined Under Namespace

Classes: RoutingStub

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket_type, options) ⇒ Engine

Returns a new instance of Engine.



16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
 
# File 'lib/omq/rust/engine.rb', line 16

def initialize(socket_type, options)
  @socket_type    = socket_type
  @options        = options
  @peer_connected = Async::Promise.new
  @all_peers_gone = Async::Promise.new
  @subscriber_joined = Async::Promise.new
  @connections    = []
  @closed         = false
  @parent_task    = nil
  @on_io_thread   = false
  @materialized   = false
  @recv_sentinels = 0
  @recv_sentinel_r = nil
  @recv_sentinel_w = nil

  @native = Native::RustSocket.new(socket_type.to_s)

  @routing = RoutingStub.new(self)
end

Instance Attribute Details

#all_peers_gone ⇒ Object (readonly)

Returns the value of attribute all_peers_gone.



9
10
11
 
# File 'lib/omq/rust/engine.rb', line 9

def all_peers_gone
  @all_peers_gone
end

#connections ⇒ Object (readonly)

Returns the value of attribute connections.



8
9
10
 
# File 'lib/omq/rust/engine.rb', line 8

def connections
  @connections
end

#on_io_thread ⇒ Object (readonly) Also known as: on_io_thread?

Returns the value of attribute on_io_thread.



10
11
12
 
# File 'lib/omq/rust/engine.rb', line 10

def on_io_thread
  @on_io_thread
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



8
9
10
 
# File 'lib/omq/rust/engine.rb', line 8

def options
  @options
end

#parent_task ⇒ Object (readonly)

Returns the value of attribute parent_task.



9
10
11
 
# File 'lib/omq/rust/engine.rb', line 9

def parent_task
  @parent_task
end

#peer_connected ⇒ Object (readonly)

Returns the value of attribute peer_connected.



9
10
11
 
# File 'lib/omq/rust/engine.rb', line 9

def peer_connected
  @peer_connected
end

#reconnect_enabled=(value) ⇒ Object (writeonly)

Sets the attribute reconnect_enabled

Parameters:

  • value —

    the value to set the attribute reconnect_enabled to.



12
13
14
 
# File 'lib/omq/rust/engine.rb', line 12

def reconnect_enabled=(value)
  @reconnect_enabled = value
end

#routing ⇒ Object (readonly)

Returns the value of attribute routing.



8
9
10
 
# File 'lib/omq/rust/engine.rb', line 8

def routing
  @routing
end

#socket_type ⇒ Object (readonly)

Returns the value of attribute socket_type.



8
9
10
 
# File 'lib/omq/rust/engine.rb', line 8

def socket_type
  @socket_type
end

#subscriber_joined ⇒ Object

Returns the value of attribute subscriber_joined.



13
14
15
 
# File 'lib/omq/rust/engine.rb', line 13

def subscriber_joined
  @subscriber_joined
end

Instance Method Details

#bind(endpoint, parent: nil) ⇒ Object 



52
53
54
55
56
57
 
# File 'lib/omq/rust/engine.rb', line 52

def bind(endpoint, parent: nil, **)
  capture_parent_task(parent: parent)
  ensure_materialized
  resolved = @native.bind(endpoint)
  URI.parse(resolved)
end

#capture_parent_task(parent: nil) ⇒ Object 



37
38
39
40
41
42
43
44
45
46
47
48
49
 
# File 'lib/omq/rust/engine.rb', line 37

def capture_parent_task(parent: nil)
  return if @parent_task

  if parent
    @parent_task = parent
  elsif Async::Task.current?
    @parent_task = Async::Task.current
  else
    @parent_task  = Reactor.root_task
    @on_io_thread = true
    Reactor.track_linger(@options.linger)
  end
end

#close ⇒ Object 



130
131
132
133
134
135
136
137
138
139
140
 
# File 'lib/omq/rust/engine.rb', line 130

def close
  return if @closed

  @closed = true
  @peer_connected.resolve(nil) unless @peer_connected.resolved?
  @all_peers_gone.resolve(nil) unless @all_peers_gone.resolved?
  @subscriber_joined.resolve(nil) unless @subscriber_joined.resolved?
  @recv_sentinel_r&.close rescue nil
  @recv_sentinel_w&.close rescue nil
  @native.close
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


143
144
145
 
# File 'lib/omq/rust/engine.rb', line 143

def closed?
  @closed
end

#connect(endpoint, parent: nil) ⇒ Object 



60
61
62
63
64
65
 
# File 'lib/omq/rust/engine.rb', line 60

def connect(endpoint, parent: nil, **)
  capture_parent_task(parent: parent)
  ensure_materialized
  @native.connect(endpoint)
  URI.parse(endpoint)
end

#dequeue_recv ⇒ Object 



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
 
# File 'lib/omq/rust/engine.rb', line 94

def dequeue_recv
  ensure_materialized

  if @recv_batch && !@recv_batch.empty?
    return @recv_batch.shift
  end

  msg = try_recv_batch
  return msg if msg

  return take_recv_sentinel if @recv_sentinels.positive?

  loop do
    readable = IO.select([@recv_signal_r, @recv_sentinel_r]).first

    if readable.include?(@recv_signal_r)
      @recv_signal_r.read_nonblock(256, exception: false)
      msg = try_recv_batch
      return msg
    end

    if readable.include?(@recv_sentinel_r)
      @recv_sentinel_r.read_nonblock(256, exception: false)
      return take_recv_sentinel if @recv_sentinels.positive?
    end
  end
end

#dequeue_recv_sentinel ⇒ Object 



123
124
125
126
127
 
# File 'lib/omq/rust/engine.rb', line 123

def dequeue_recv_sentinel
  @recv_sentinels += 1
  @recv_sentinel_w&.write_nonblock(".", exception: false) rescue nil
  nil
end

#disconnect(endpoint) ⇒ Object 



68
69
70
 
# File 'lib/omq/rust/engine.rb', line 68

def disconnect(endpoint)
  @native.disconnect(endpoint)
end

#emit_monitor_event(type, endpoint: nil, detail: nil) ⇒ Object 



158
159
 
# File 'lib/omq/rust/engine.rb', line 158

def emit_monitor_event(type, endpoint: nil, detail: nil)
end

#enqueue_send(parts) ⇒ Object 



78
79
80
81
82
83
84
85
86
87
88
89
90
91
 
# File 'lib/omq/rust/engine.rb', line 78

def enqueue_send(parts)
  ensure_materialized
  result = @native.enqueue_send(parts)
  return if result == :ok

  @send_signal_r ||= IO.for_fd(@native.send_fd, autoclose: false)
  loop do
    result = @native.enqueue_send(parts)
    return if result == :ok

    @send_signal_r.wait_readable
    @send_signal_r.read_nonblock(256, exception: false)
  end
end

#monitor_queue=(queue) ⇒ Object 



162
163
164
165
166
167
 
# File 'lib/omq/rust/engine.rb', line 162

def monitor_queue=(queue)
  @monitor_queue = queue
  return unless queue && @materialized

  start_monitor_forwarder
end

#subscribe(prefix) ⇒ Object 



148
149
150
 
# File 'lib/omq/rust/engine.rb', line 148

def subscribe(prefix)
  @routing.subscribe(prefix)
end

#unbind(endpoint) ⇒ Object 



73
74
75
 
# File 'lib/omq/rust/engine.rb', line 73

def unbind(endpoint)
  @native.unbind(endpoint)
end

#unsubscribe(prefix) ⇒ Object 



153
154
155
 
# File 'lib/omq/rust/engine.rb', line 153

def unsubscribe(prefix)
  @routing.unsubscribe(prefix)
end

#verbose_monitor=(val) ⇒ Object 



170
171
172
 
# File 'lib/omq/rust/engine.rb', line 170

def verbose_monitor=(val)
  @verbose_monitor = val
end