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.



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 17

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
  @compression_options  = {}

  @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.



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

def all_peers_gone
  @all_peers_gone
end

#connections ⇒ Object (readonly)

Returns the value of attribute connections.



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

def connections
  @connections
end

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

Returns the value of attribute on_io_thread.



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

def on_io_thread
  @on_io_thread
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



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

def options
  @options
end

#parent_task ⇒ Object (readonly)

Returns the value of attribute parent_task.



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

def parent_task
  @parent_task
end

#peer_connected ⇒ Object (readonly)

Returns the value of attribute peer_connected.



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

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.



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

def reconnect_enabled=(value)
  @reconnect_enabled = value
end

#routing ⇒ Object (readonly)

Returns the value of attribute routing.



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

def routing
  @routing
end

#socket_type ⇒ Object (readonly)

Returns the value of attribute socket_type.



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

def socket_type
  @socket_type
end

#subscriber_joined ⇒ Object

Returns the value of attribute subscriber_joined.



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

def subscriber_joined
  @subscriber_joined
end

Instance Method Details

#bind(endpoint, parent: nil, **opts) ⇒ Object 



54
55
56
57
58
59
60
 
# File 'lib/omq/rust/engine.rb', line 54

def bind(endpoint, parent: nil, **opts)
  capture_parent_task(parent: parent)
  apply_endpoint_options!(opts)
  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
50
51
 
# 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
  elsif !Reactor.native_fiber_scheduler?
    @parent_task = nil
  else
    @parent_task  = Reactor.root_task
    @on_io_thread = true
    Reactor.track_linger(@options.linger)
  end
end

#close ⇒ Object 



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

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?
  FdWatcher.unwatch_owner(self) unless Reactor.native_fiber_scheduler?
  @native.close
  close_io_wrapper(@recv_signal_r)
  close_io_wrapper(@send_signal_r)
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


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

def closed?
  @closed
end

#connect(endpoint, parent: nil, **opts) ⇒ Object 



63
64
65
66
67
68
69
 
# File 'lib/omq/rust/engine.rb', line 63

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

#dequeue_recv ⇒ Object 



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

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 



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

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

#disconnect(endpoint) ⇒ Object 



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

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 



82
83
84
85
86
87
88
89
90
91
92
93
94
95
 
# File 'lib/omq/rust/engine.rb', line 82

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

  @send_signal_r ||= io_for_native_fd(@native.send_fd)
  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 



77
78
79
 
# File 'lib/omq/rust/engine.rb', line 77

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