Class: OMQ::Ractor::SocketProxy

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

Overview

Ractor-side proxy for an OMQ socket. Provides #receive, #<<, and #publish to communicate with the main-thread socket through Ractor ports.

Instance Method Summary collapse

Constructor Details

#initialize(input_port, output_port, topic_type) ⇒ SocketProxy

Returns a new instance of SocketProxy.

Parameters:

  • input_port (Ractor::Port, nil) —

    port for receiving messages (nil if write-only)

  • output_port (Ractor::Port, nil) —

    port for sending messages (nil if read-only)

  • topic_type (Boolean) —

    whether this is a topic-based socket (PUB/SUB, RADIO/DISH)



209
210
211
212
213
214
 
# File 'lib/omq/ractor.rb', line 209

def initialize(input_port, output_port, topic_type)
  @in         = input_port
  @out        = output_port
  @topic_type = topic_type
  @closed     = false
end

Instance Method Details

#<<(msg) ⇒ self

Sends a message through this socket. For topic-based sockets, wraps as ["", obj] (all subscribers).

Parameters:

  • msg (Object) —

    message

Returns:

  • (self)

Raises:

  • (::Ractor::ClosedError) 


257
258
259
260
261
262
263
264
265
 
# File 'lib/omq/ractor.rb', line 257

def <<(msg)
  raise ::Ractor::ClosedError, "not writable" unless @out
  if @topic_type
    @out.send(["".b.freeze, msg])
  else
    @out.send(msg)
  end
  self
end

#publish(msg, topic:) ⇒ self

Publishes a message with an explicit topic (PUB/SUB, RADIO/DISH).

Parameters:

  • msg (Object) —

    payload

  • topic (String) —

    topic string for subscription matching

Returns:

  • (self)

Raises:

  • (::Ractor::ClosedError) 


274
275
276
277
278
 
# File 'lib/omq/ractor.rb', line 274

def publish(msg, topic:)
  raise ::Ractor::ClosedError, "not writable" unless @out
  @out.send([topic.b.freeze, msg])
  self
end

#receive ⇒ Object?

Receives the next message from this socket. Returns nil once when the socket closes, then raises SocketClosedError on subsequent calls.

Returns:

  • (Object, nil) —

    deserialized message, or nil on close

Raises:

  • (::Ractor::ClosedError) 


223
224
225
226
227
228
229
230
231
232
 
# File 'lib/omq/ractor.rb', line 223

def receive
  raise ::Ractor::ClosedError, "not readable" unless @in
  raise SocketClosedError, "socket closed" if @closed
  msg = @in.receive
  if msg.nil?
    @closed = true
    return nil
  end
  @topic_type ? msg.last : msg
end

#receive_with_topic ⇒ Array(String, Object)?

Receives the next message with its topic (PUB/SUB, RADIO/DISH).

Returns:

  • (Array(String, Object), nil) —

    [topic, payload], or nil on close

Raises:

  • (::Ractor::ClosedError) 


239
240
241
242
243
244
245
246
247
248
 
# File 'lib/omq/ractor.rb', line 239

def receive_with_topic
  raise ::Ractor::ClosedError, "not readable" unless @in
  raise SocketClosedError, "socket closed" if @closed
  msg = @in.receive
  if msg.nil?
    @closed = true
    return nil
  end
  [msg.first, msg.last]
end

#to_port ⇒ Ractor::Port

Returns the input port for use with Ractor.select.

Returns:

  • (Ractor::Port)

Raises:

  • (::Ractor::ClosedError) 


285
286
287
288
 
# File 'lib/omq/ractor.rb', line 285

def to_port
  raise ::Ractor::ClosedError, "not readable" unless @in
  @in
end