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
 
# 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

  @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 



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

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 



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

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 



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

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?
  @native.close
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


134
135
136
 
# File 'lib/omq/rust/engine.rb', line 134

def closed?
  @closed
end

#connect(endpoint, parent: nil) ⇒ Object 



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

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

#dequeue_recv ⇒ Object 



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
 
# File 'lib/omq/rust/engine.rb', line 92

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
    @recv_signal_r.wait_readable
    @recv_signal_r.read_nonblock(256, exception: false)

    msg = try_recv_batch
    return msg if msg

    return take_recv_sentinel if @recv_sentinels.positive?
  end
end

#dequeue_recv_sentinel ⇒ Object 



116
117
118
119
120
 
# File 'lib/omq/rust/engine.rb', line 116

def dequeue_recv_sentinel
  @recv_sentinels += 1
  @native.wake_recv if @materialized
  nil
end

#disconnect(endpoint) ⇒ Object 



66
67
68
 
# File 'lib/omq/rust/engine.rb', line 66

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

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



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

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

#enqueue_send(parts) ⇒ Object 



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

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 



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

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

  start_monitor_forwarder
end

#subscribe(prefix) ⇒ Object 



139
140
141
 
# File 'lib/omq/rust/engine.rb', line 139

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

#unbind(endpoint) ⇒ Object 



71
72
73
 
# File 'lib/omq/rust/engine.rb', line 71

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

#unsubscribe(prefix) ⇒ Object 



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

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

#verbose_monitor=(val) ⇒ Object 



161
162
163
 
# File 'lib/omq/rust/engine.rb', line 161

def verbose_monitor=(val)
  @verbose_monitor = val
end