Module: KubeMQ::Transport::Converter Private

Defined in:
lib/kubemq/transport/converter.rb

Overview

This module is part of a private API. You should avoid using this module if possible, as it may be removed or be changed in the future.

Bidirectional conversions between SDK domain objects and gRPC protobuf messages.

All methods are module-level (+module_function+) and stateless. They handle body encoding (binary-safe), tag validation, UUID generation for missing IDs, and subscription type routing.

rubocop:disable Metrics/ModuleLength -- bidirectional proto mapping helpers

See Also:

Class Method Summary collapse

Class Method Details

.encode_body(body) ⇒ String

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Encodes a message body to binary (ASCII-8BIT).

Parameters:

  • body (String, nil) —

    the message body to encode

Returns:

  • (String) —

    binary-encoded body (empty binary string if nil)

Raises:

  • (ArgumentError) —

    if body is not a String or nil



278
279
280
281
282
283
 
# File 'lib/kubemq/transport/converter.rb', line 278

def encode_body(body)
  return ''.b if body.nil?
  raise ArgumentError, "body must be a String, got #{body.class}" unless body.is_a?(String)

  body.dup.force_encoding('ASCII-8BIT')
end

.event_to_proto(message, client_id, store: false) ⇒ Kubemq::Event

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Converts an SDK event message to a protobuf Kubemq::Event.

Parameters:

  • message (PubSub::EventMessage, PubSub::EventStoreMessage) —

    the outbound event

  • client_id (String) —

    the client identifier

  • store (Boolean) (defaults to: false) —

    true for events store (durable), false for fire-and-forget

Returns:

Raises:



29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
 
# File 'lib/kubemq/transport/converter.rb', line 29

def event_to_proto(message, client_id, store: false)
  body = encode_body(message.body)
  tags = message.respond_to?(:tags) ? (message.tags || {}) : {}
  validate_tags!(tags)

  ::Kubemq::Event.new(
    EventID: message.id || SecureRandom.uuid,
    ClientID: client_id,
    Channel: message.channel,
    Metadata: message.metadata || '',
    Body: body,
    Store: store,
    Tags: tags
  )
end

.proto_to_command_response(response) ⇒ Hash{Symbol => Object}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extracts command response fields from a protobuf Kubemq::Response.

Parameters:

Returns:

  • (Hash{Symbol => Object}) —

    :client_id, :request_id, :executed, :timestamp, :error, :tags



205
206
207
208
209
210
211
212
213
214
 
# File 'lib/kubemq/transport/converter.rb', line 205

def proto_to_command_response(response)
  {
    client_id: response.ClientID,
    request_id: response.RequestID,
    executed: response.Executed,
    timestamp: response.Timestamp,
    error: response.Error.empty? ? nil : response.Error,
    tags: response.Tags.to_h
  }
end

.proto_to_event_received(event_receive) ⇒ Hash{Symbol => Object}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extracts received event fields from a protobuf Kubemq::EventReceive.

Parameters:

Returns:

  • (Hash{Symbol => Object}) —

    :id, :channel, :metadata, :body, :timestamp, :sequence, :tags



62
63
64
65
66
67
68
69
70
71
72
 
# File 'lib/kubemq/transport/converter.rb', line 62

def proto_to_event_received(event_receive)
  {
    id: event_receive.EventID,
    channel: event_receive.Channel,
    metadata: event_receive.Metadata,
    body: event_receive.Body,
    timestamp: event_receive.Timestamp,
    sequence: event_receive.Sequence,
    tags: event_receive.Tags.to_h
  }
end

.proto_to_event_result(result) ⇒ Hash{Symbol => Object}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extracts send result fields from a protobuf Kubemq::Result.

Parameters:

Returns:

  • (Hash{Symbol => Object}) —

    :id, :sent, :error fields



49
50
51
52
53
54
55
 
# File 'lib/kubemq/transport/converter.rb', line 49

def proto_to_event_result(result)
  {
    id: result.EventID,
    sent: result.Sent,
    error: result.Error.empty? ? nil : result.Error
  }
end

.proto_to_query_response(response) ⇒ Hash{Symbol => Object}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extracts query response fields from a protobuf Kubemq::Response.

Parameters:

Returns:

  • (Hash{Symbol => Object}) —

    :client_id, :request_id, :executed, :metadata, :body, :cache_hit, :timestamp, :error, :tags



221
222
223
224
225
226
227
228
229
230
231
232
233
 
# File 'lib/kubemq/transport/converter.rb', line 221

def proto_to_query_response(response)
  {
    client_id: response.ClientID,
    request_id: response.RequestID,
    executed: response.Executed,
    metadata: response.Metadata,
    body: response.Body,
    cache_hit: response.CacheHit,
    timestamp: response.Timestamp,
    error: response.Error.empty? ? nil : response.Error,
    tags: response.Tags.to_h
  }
end

.proto_to_queue_message_received(msg) ⇒ Hash{Symbol => Object}

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Extracts received queue message fields from a protobuf Kubemq::QueueMessage.

Parameters:

Returns:

  • (Hash{Symbol => Object}) —

    :id, :channel, :metadata, :body, :tags, :attributes



150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
 
# File 'lib/kubemq/transport/converter.rb', line 150

def proto_to_queue_message_received(msg)
  attrs = if msg.Attributes
            {
              timestamp: msg.Attributes.Timestamp,
              sequence: msg.Attributes.Sequence,
              md5_of_body: msg.Attributes.MD5OfBody,
              receive_count: msg.Attributes.ReceiveCount,
              re_routed: msg.Attributes.ReRouted,
              re_routed_from_queue: msg.Attributes.ReRoutedFromQueue,
              expiration_at: msg.Attributes.ExpirationAt,
              delayed_to: msg.Attributes.DelayedTo
            }
          end

  {
    id: msg.MessageID,
    channel: msg.Channel,
    metadata: msg.Metadata,
    body: msg.Body,
    tags: msg.Tags.to_h,
    attributes: attrs
  }
end

.queue_message_to_proto(message, client_id) ⇒ Kubemq::QueueMessage

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Converts an SDK queue message to a protobuf Kubemq::QueueMessage.

Includes the optional delivery policy (expiration, delay, dead letter) when present on the message.

Parameters:

  • message (Queues::QueueMessage) —

    the outbound queue message

  • client_id (String) —

    the client identifier

Returns:

Raises:



118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
 
# File 'lib/kubemq/transport/converter.rb', line 118

def queue_message_to_proto(message, client_id)
  body = encode_body(message.body)
  tags = message.respond_to?(:tags) ? (message.tags || {}) : {}
  validate_tags!(tags)

  msg = ::Kubemq::QueueMessage.new(
    MessageID: message.id || SecureRandom.uuid,
    ClientID: client_id,
    Channel: message.channel,
    Metadata: message.metadata || '',
    Body: body,
    Tags: tags
  )

  if message.respond_to?(:policy) && message.policy
    policy = message.policy
    msg.Policy = ::Kubemq::QueueMessagePolicy.new(
      ExpirationSeconds: policy.expiration_seconds || 0,
      DelaySeconds: policy.delay_seconds || 0,
      MaxReceiveCount: policy.max_receive_count || 0,
      MaxReceiveQueue: policy.max_receive_queue || ''
    )
  end

  msg
end

.request_to_proto(message, client_id, type) ⇒ Kubemq::Request

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Converts an SDK command or query message to a protobuf Kubemq::Request.

Parameters:

Returns:

Raises:



181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
 
# File 'lib/kubemq/transport/converter.rb', line 181

def request_to_proto(message, client_id, type)
  body = encode_body(message.body)
  tags = message.respond_to?(:tags) ? (message.tags || {}) : {}
  validate_tags!(tags)

  ::Kubemq::Request.new(
    RequestID: message.id || SecureRandom.uuid,
    RequestTypeData: type,
    ClientID: client_id,
    Channel: message.channel,
    Metadata: message.metadata || '',
    Body: body,
    Timeout: message.timeout || 10_000,
    CacheKey: message.respond_to?(:cache_key) ? (message.cache_key || '') : '',
    CacheTTL: message.respond_to?(:cache_ttl) ? (message.cache_ttl || 0) : 0,
    Tags: tags
  )
end

.response_message_to_proto(response, client_id) ⇒ Kubemq::Response

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Converts an SDK response message to a protobuf Kubemq::Response.

Parameters:

Returns:

Raises:



241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
 
# File 'lib/kubemq/transport/converter.rb', line 241

def response_message_to_proto(response, client_id)
  raw_body = response.respond_to?(:body) ? response.body : nil
  body = encode_body(raw_body)
  tags = response.respond_to?(:tags) ? (response.tags || {}) : {}
  validate_tags!(tags)

  ::Kubemq::Response.new(
    ClientID: client_id,
    RequestID: response.request_id,
    ReplyChannel: response.reply_channel,
    Metadata: response.respond_to?(:metadata) ? (response.metadata || '') : '',
    Body: body,
    Executed: response.respond_to?(:executed) ? response.executed : true,
    Error: response.respond_to?(:error) ? (response.error || '') : '',
    Tags: tags
  )
end

.subscribe_to_proto(subscription, client_id) ⇒ Kubemq::Subscribe

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

Converts an SDK subscription object to a protobuf Kubemq::Subscribe.

Routes to the correct SubscribeTypeData based on SubscribeType and populates events store start position fields when applicable.

Parameters:

Returns:

Raises:

  • (ArgumentError) —

    if the subscribe type is unknown



84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
 
# File 'lib/kubemq/transport/converter.rb', line 84

def subscribe_to_proto(subscription, client_id)
  sub = ::Kubemq::Subscribe.new(
    ClientID: client_id,
    Channel: subscription.channel,
    Group: subscription.respond_to?(:group) ? (subscription.group || '') : ''
  )

  case subscription.subscribe_type
  when SubscribeType::EVENTS
    sub.SubscribeTypeData = SubscribeType::EVENTS
  when SubscribeType::EVENTS_STORE
    sub.SubscribeTypeData = SubscribeType::EVENTS_STORE
    sub.EventsStoreTypeData = subscription.start_position || 0
    sub.EventsStoreTypeValue = subscription.start_position_value || 0
  when SubscribeType::COMMANDS
    sub.SubscribeTypeData = SubscribeType::COMMANDS
  when SubscribeType::QUERIES
    sub.SubscribeTypeData = SubscribeType::QUERIES
  else
    raise ArgumentError, "Unknown subscribe_type: #{subscription.subscribe_type.inspect}"
  end

  sub
end

.validate_tags!(tags) ⇒ void

This method is part of a private API. You should avoid using this method if possible, as it may be removed or be changed in the future.

This method returns an undefined value.

Validates that all tag keys and values are strings.

Parameters:

  • tags (Hash, nil) —

    key-value tag map to validate

Raises:



264
265
266
267
268
269
270
271
 
# File 'lib/kubemq/transport/converter.rb', line 264

def validate_tags!(tags)
  return if tags.nil?

  tags.each do |k, v|
    raise ValidationError, "Tag key must be String, got #{k.class}" unless k.is_a?(String)
    raise ValidationError, "Tag value must be String, got #{v.class} for key '#{k}'" unless v.is_a?(String)
  end
end