Module: KubeMQ::Transport::ChannelManager Private

Defined in:
lib/kubemq/transport/channel_manager.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.

Channel lifecycle operations (create, delete, list, purge) sent as internal queries over the KubeMQ cluster management channel.

All methods are module-level (+module_function+) and delegate the underlying gRPC call to the provided GrpcTransport instance.

rubocop:disable Metrics/ModuleLength -- internal channel CRUD + list helpers

See Also:

Constant Summary collapse

INTERNAL_CHANNEL =

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

Cluster-internal channel used for management requests.

'kubemq.cluster.internal.requests'
INTERNAL_TIMEOUT =

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

Default timeout (in milliseconds) for management requests.

10_000

Class Method Summary collapse

Class Method Details

.build_request(client_id:, metadata:, tags:) ⇒ 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.

Builds a protobuf management request targeting the internal channel.

Parameters:

  • client_id (String) —

    the client identifier

  • metadata (String) —

    management operation name

  • tags (Hash{String => String}) —

    operation parameters as tags

Returns:



165
166
167
168
169
170
171
172
173
174
175
 
# File 'lib/kubemq/transport/channel_manager.rb', line 165

def build_request(client_id:, metadata:, tags:)
  ::Kubemq::Request.new(
    RequestID: SecureRandom.uuid,
    RequestTypeData: RequestType::QUERY,
    ClientID: client_id,
    Channel: INTERNAL_CHANNEL,
    Metadata: metadata,
    Timeout: INTERNAL_TIMEOUT,
    Tags: tags
  )
end

.check_response_error!(response, operation, channel = nil) ⇒ 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.

Raises ChannelError if the response contains a non-empty error string.

Parameters:

  • response (Kubemq::Response) —

    protobuf response to check

  • operation (String) —

    the operation name for error context

  • channel (String, nil) (defaults to: nil) —

    the channel name for error context

Raises:



184
185
186
187
188
189
190
191
192
193
 
# File 'lib/kubemq/transport/channel_manager.rb', line 184

def check_response_error!(response, operation, channel = nil)
  return if response.Error.nil? || response.Error.empty?

  raise ChannelError.new(
    "#{operation} failed: #{response.Error}",
    code: ErrorCode::INTERNAL,
    operation: operation,
    channel: channel
  )
end

.create_channel(transport, client_id, channel_name, channel_type) ⇒ Boolean

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.

Creates a channel on the KubeMQ broker.

rubocop:disable Naming/PredicateMethod -- command-style API returns true on success

Parameters:

  • transport (GrpcTransport) —

    the active transport instance

  • client_id (String) —

    the client identifier

  • channel_name (String) —

    name for the new channel

  • channel_type (String) —

    one of ChannelType constants

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
 
# File 'lib/kubemq/transport/channel_manager.rb', line 40

def create_channel(transport, client_id, channel_name, channel_type)
  request = build_request(
    client_id: client_id,
    metadata: 'create-channel',
    tags: {
      'channel_type' => channel_type,
      'channel' => channel_name,
      'client_id' => client_id
    }
  )

  response = transport.kubemq_client.send_request(request)
  check_response_error!(response, 'create_channel', channel_name)
  true
end

.delete_channel(transport, client_id, channel_name, channel_type) ⇒ Boolean

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.

Deletes a channel from the KubeMQ broker.

rubocop:disable Naming/PredicateMethod -- command-style API returns true on success

Parameters:

  • transport (GrpcTransport) —

    the active transport instance

  • client_id (String) —

    the client identifier

  • channel_name (String) —

    name of the channel to delete

  • channel_type (String) —

    one of ChannelType constants

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



68
69
70
71
72
73
74
75
76
77
78
79
80
81
 
# File 'lib/kubemq/transport/channel_manager.rb', line 68

def delete_channel(transport, client_id, channel_name, channel_type)
  request = build_request(
    client_id: client_id,
    metadata: 'delete-channel',
    tags: {
      'channel_type' => channel_type,
      'channel' => channel_name
    }
  )

  response = transport.kubemq_client.send_request(request)
  check_response_error!(response, 'delete_channel', channel_name)
  true
end

.list_channels(transport, client_id, channel_type, search = nil) ⇒ Array<ChannelInfo>

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.

Lists channels of the specified type, with optional name filtering.

Retries up to 3 times when the cluster snapshot is not ready.

Parameters:

  • transport (GrpcTransport) —

    the active transport instance

  • client_id (String) —

    the client identifier

  • channel_type (String) —

    one of ChannelType constants

  • search (String, nil) (defaults to: nil) —

    substring filter for channel names

Returns:

  • (Array<ChannelInfo>) —

    matching channels with metadata

Raises:

  • (ChannelError) —

    if the broker rejects the operation after retries

See Also:



96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
 
# File 'lib/kubemq/transport/channel_manager.rb', line 96

def list_channels(transport, client_id, channel_type, search = nil)
  tags = { 'channel_type' => channel_type }
  tags['channel_search'] = search if search && !search.empty?

  request = build_request(
    client_id: client_id,
    metadata: 'list-channels',
    tags: tags
  )

  max_retries = 3
  response = nil
  max_retries.times do |attempt|
    response = transport.kubemq_client.send_request(request)

    raise StandardError, response.Error if response.Error&.include?('cluster snapshot not ready')

    break
  rescue StandardError => e
    if e.message.include?('cluster snapshot not ready') && attempt < max_retries - 1
      sleep(1)
      next
    end
    raise ChannelError.new(
      "list_channels failed after #{attempt + 1} attempts: #{e.message}",
      operation: 'list_channels'
    )
  end

  check_response_error!(response, 'list_channels')
  parse_channel_list(response.Body, channel_type)
end

.parse_channel_list(body, channel_type) ⇒ Array<ChannelInfo>

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.

Parses a JSON channel list response body into ChannelInfo objects.

Parameters:

  • body (String, nil) —

    JSON response body

  • channel_type (String) —

    channel type to assign to each entry

Returns:



200
201
202
203
204
205
206
207
208
 
# File 'lib/kubemq/transport/channel_manager.rb', line 200

def parse_channel_list(body, channel_type)
  return [] if body.nil? || body.empty?

  data = JSON.parse(body.dup.force_encoding('UTF-8'))
  channels = data.is_a?(Array) ? data : (data['channels'] || data['items'] || [data])
  channels.compact.map { |ch| ChannelInfo.from_json(ch.merge('type' => channel_type)) }
rescue JSON::ParserError
  []
end

.purge_queue(transport, client_id, channel_name) ⇒ Hash{Symbol => Integer}

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.

Purges all pending messages from a queue channel.

Parameters:

  • transport (GrpcTransport) —

    the active transport instance

  • client_id (String) —

    the client identifier

  • channel_name (String) —

    queue channel to purge

Returns:

  • (Hash{Symbol => Integer}) —

    { affected_messages: Integer }

Raises:

See Also:



138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
 
# File 'lib/kubemq/transport/channel_manager.rb', line 138

def purge_queue(transport, client_id, channel_name)
  request = ::Kubemq::AckAllQueueMessagesRequest.new(
    RequestID: SecureRandom.uuid,
    ClientID: client_id,
    Channel: channel_name,
    WaitTimeSeconds: 5
  )

  response = transport.kubemq_client.ack_all_queue_messages(request)
  if response.IsError && !response.Error.empty?
    raise ChannelError.new(
      "Failed to purge queue '#{channel_name}': #{response.Error}",
      code: ErrorCode::INTERNAL,
      operation: 'purge_queue',
      channel: channel_name
    )
  end

  { affected_messages: response.AffectedMessages }
end