Class: OMQ::Rust::Java::Engine

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

Defined Under Namespace

Classes: RoutingStub

Constant Summary collapse

POLL_SECONDS = 
0.005
POLL_DURATION = 
Duration.ofMillis((POLL_SECONDS * 1000).to_i)

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(socket_type, options) ⇒ Engine

Returns a new instance of Engine.



88
89
90
91
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/java/engine.rb', line 88

def initialize(socket_type, options)
  @socket_type         = socket_type
  @options             = options
  @connections         = {}
  @closed              = false
  @parent_task         = nil
  @on_io_thread        = false
  @materialized        = false
  @recv_sentinels      = 0
  @compression_options = {}

  @peer_connected = Promise.new do |promise|
    wait_connected(1)
    promise.resolve(true) unless @closed
  end
  @all_peers_gone = Promise.new do |promise|
    wait_all_peers_gone
    promise.resolve(true) unless @closed
  end
  @subscriber_joined = Promise.new do |promise|
    wait_subscribed(1)
    promise.resolve(true) unless @closed
  end

  @routing = RoutingStub.new(self)
end

Instance Attribute Details

#all_peers_gone ⇒ Object (readonly)

Returns the value of attribute all_peers_gone.



81
82
83
 
# File 'lib/omq/rust/java/engine.rb', line 81

def all_peers_gone
  @all_peers_gone
end

#connections ⇒ Object (readonly)

Returns the value of attribute connections.



80
81
82
 
# File 'lib/omq/rust/java/engine.rb', line 80

def connections
  @connections
end

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

Returns the value of attribute on_io_thread.



82
83
84
 
# File 'lib/omq/rust/java/engine.rb', line 82

def on_io_thread
  @on_io_thread
end

#options ⇒ Object (readonly)

Returns the value of attribute options.



80
81
82
 
# File 'lib/omq/rust/java/engine.rb', line 80

def options
  @options
end

#parent_task ⇒ Object (readonly)

Returns the value of attribute parent_task.



81
82
83
 
# File 'lib/omq/rust/java/engine.rb', line 81

def parent_task
  @parent_task
end

#peer_connected ⇒ Object (readonly)

Returns the value of attribute peer_connected.



81
82
83
 
# File 'lib/omq/rust/java/engine.rb', line 81

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.



84
85
86
 
# File 'lib/omq/rust/java/engine.rb', line 84

def reconnect_enabled=(value)
  @reconnect_enabled = value
end

#routing ⇒ Object (readonly)

Returns the value of attribute routing.



80
81
82
 
# File 'lib/omq/rust/java/engine.rb', line 80

def routing
  @routing
end

#socket_type ⇒ Object (readonly)

Returns the value of attribute socket_type.



80
81
82
 
# File 'lib/omq/rust/java/engine.rb', line 80

def socket_type
  @socket_type
end

#subscriber_joined ⇒ Object

Returns the value of attribute subscriber_joined.



85
86
87
 
# File 'lib/omq/rust/java/engine.rb', line 85

def subscriber_joined
  @subscriber_joined
end

Instance Method Details

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



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

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

#capture_parent_task(parent: nil) ⇒ Object 



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

def capture_parent_task(parent: nil)
  @parent_task ||= parent
end

#close ⇒ Object Also known as: stop 



191
192
193
194
195
196
197
198
199
200
201
 
# File 'lib/omq/rust/java/engine.rb', line 191

def close
  return if @closed

  @closed = true
  @peer_connected.resolve(nil)
  @all_peers_gone.resolve(nil)
  @subscriber_joined.resolve(nil)
  with_java_errors { @native&.close }
  @connections.clear
  nil
end

#closed? ⇒ Boolean

Returns:

  • (Boolean) 


207
208
209
 
# File 'lib/omq/rust/java/engine.rb', line 207

def closed?
  @closed
end

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



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

def connect(endpoint, parent: nil, **opts)
  capture_parent_task(parent: parent)
  apply_endpoint_options!(opts)
  ensure_materialized
  with_java_errors { @native.connect(endpoint) }
  resolve_peer_connected if endpoint.start_with?("inproc://")
  URI.parse(endpoint)
end

#dequeue_recv ⇒ Object 



166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
 
# File 'lib/omq/rust/java/engine.rb', line 166

def dequeue_recv
  ensure_materialized
  @recv_deadline = monotonic_time + @options.read_timeout.to_f if @options.read_timeout

  loop do
    return take_recv_sentinel if @recv_sentinels.positive?

    optional = with_java_errors { @native.tryReceive }
    return ruby_parts(optional.get) if optional.isPresent

    raise IO::TimeoutError, "operation timed out" if recv_deadline_expired?

    sleep recv_poll_seconds
  end
ensure
  @recv_deadline = nil
end

#dequeue_recv_sentinel ⇒ Object 



185
186
187
188
 
# File 'lib/omq/rust/java/engine.rb', line 185

def dequeue_recv_sentinel
  @recv_sentinels += 1
  nil
end

#disconnect(endpoint) ⇒ Object 



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

def disconnect(endpoint)
  ensure_materialized
  with_java_errors { @native.disconnect(endpoint) }
end

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



222
223
 
# File 'lib/omq/rust/java/engine.rb', line 222

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

#enqueue_send(parts) ⇒ Object 



151
152
153
154
155
156
157
158
159
160
161
162
163
 
# File 'lib/omq/rust/java/engine.rb', line 151

def enqueue_send(parts)
  ensure_materialized
  msg = java_message(parts)

  if (timeout = @options.write_timeout)
    ok = with_java_errors { @native.send(msg, duration_from_seconds("write_timeout", timeout)) }
    raise IO::TimeoutError, "operation timed out" unless ok
  else
    with_java_errors { @native.send(msg) }
  end

  nil
end

#monitor_queue=(queue) ⇒ Object 



226
227
228
 
# File 'lib/omq/rust/java/engine.rb', line 226

def monitor_queue=(queue)
  @monitor_queue = queue
end

#subscribe(prefix) ⇒ Object 



212
213
214
 
# File 'lib/omq/rust/java/engine.rb', line 212

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

#unbind(endpoint) ⇒ Object 



145
146
147
148
 
# File 'lib/omq/rust/java/engine.rb', line 145

def unbind(endpoint)
  ensure_materialized
  with_java_errors { @native.unbind(endpoint) }
end

#unsubscribe(prefix) ⇒ Object 



217
218
219
 
# File 'lib/omq/rust/java/engine.rb', line 217

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

#verbose_monitor=(val) ⇒ Object 



231
232
233
 
# File 'lib/omq/rust/java/engine.rb', line 231

def verbose_monitor=(val)
  @verbose_monitor = val
end