Class: OMQ::Ractor::TopicMarshalConnection

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

Overview

Topic-aware Marshal wrapper for PUB/XPUB/RADIO (send side). Preserves parts (topic/group) as a plain string for subscription matching; serializes parts (payload).

Instance Method Summary collapse

Methods included from TransparentDelegator

#is_a?

Constructor Details

#initialize(conn, cache) ⇒ TopicMarshalConnection

Returns a new instance of TopicMarshalConnection.

Parameters:

  • conn (Object) —

    underlying connection to wrap

  • cache (SerializeCache) —

    shared serialization cache



146
147
148
149
 
# File 'lib/omq/ractor.rb', line 146

def initialize(conn, cache)
  super(conn)
  @cache = cache
end

Instance Method Details

#receive_message ⇒ Array<String>

Returns deserialized message with topic as first element.

Returns:

  • (Array<String>) —

    deserialized message with topic as first element



174
175
176
177
 
# File 'lib/omq/ractor.rb', line 174

def receive_message
  parts = super
  [parts[0], *Marshal.load(parts[1])]
end

#send_message(parts) ⇒ void

This method returns an undefined value.

Parameters:

  • parts (Array<String>) —

    message frames; first frame is topic



154
155
156
 
# File 'lib/omq/ractor.rb', line 154

def send_message(parts)
  super([parts[0], @cache.marshal(parts[1..])])
end

#write_message(parts) ⇒ void

This method returns an undefined value.

Parameters:

  • parts (Array<String>) —

    message frames; first frame is topic



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

def write_message(parts)
  super([parts[0], @cache.marshal(parts[1..])])
end

#write_messages(batch) ⇒ void

This method returns an undefined value.

Parameters:

  • batch (Array<Array<String>>) —

    batch of message frames; first frame is topic



168
169
170
 
# File 'lib/omq/ractor.rb', line 168

def write_messages(batch)
  super(batch.map { |parts| [parts[0], @cache.marshal(parts[1..])] })
end