Class: KubeMQ::BaseClient

Inherits:
Object
  • Object
show all
Defined in:
lib/kubemq/base_client.rb

Overview

Note:

This class is thread-safe. The underlying transport and state are protected by a mutex.

Base class for all KubeMQ client types.

Provides shared connection management, channel CRUD operations, and lifecycle methods. Not intended for direct instantiation — use PubSubClient, QueuesClient, or CQClient instead.

See Also:

Direct Known Subclasses

CQClient, PubSubClient, QueuesClient

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(address: nil, client_id: nil, auth_token: nil, config: nil, transport: nil, **options) ⇒ BaseClient

Creates a new client connected to a KubeMQ broker.

Connection is established lazily on first use, not during initialization. Pass individual options or a pre-built Configuration object.

Examples:

client = KubeMQ::PubSubClient.new(
  address: "localhost:50000",
  client_id: "my-publisher"
)

Parameters:

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

    broker host:port (default: from config or localhost:50000)

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

    unique identifier for this client (auto-generated if nil)

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

    bearer token for authentication

  • config (Configuration, nil) (defaults to: nil) —

    pre-built configuration (overrides individual options)

  • transport (Transport::GrpcTransport, nil) (defaults to: nil) —

    custom transport (for testing)

  • options (Hash) —

    additional options forwarded to Configuration

Raises:



44
45
46
47
48
49
50
51
52
53
54
55
 
# File 'lib/kubemq/base_client.rb', line 44

def initialize(address: nil, client_id: nil, auth_token: nil, config: nil, transport: nil, **options)
  @config = config || Configuration.new(
    address: address,
    client_id: client_id,
    auth_token: auth_token,
    **options
  )
  @config.validate!
  @transport = transport || Transport::GrpcTransport.new(@config)
  @closed = false
  @mutex = Mutex.new
end

Instance Attribute Details

#config ⇒ Configuration (readonly)

Returns the client's resolved configuration.

Returns:



23
24
25
 
# File 'lib/kubemq/base_client.rb', line 23

def config
  @config
end

#transport ⇒ Object (readonly)

Returns the value of attribute transport.



23
 
# File 'lib/kubemq/base_client.rb', line 23

attr_reader :config, :transport

Instance Method Details

#close ⇒ void

This method returns an undefined value.

Closes the client and releases all resources.

Cancels active subscriptions, flushes the message buffer, and closes the gRPC channel. This method is idempotent — calling it multiple times is safe.



79
80
81
82
 
# File 'lib/kubemq/base_client.rb', line 79

def close
  @mutex.synchronize { @closed = true }
  @transport.close
end

#closed? ⇒ Boolean

Returns whether the client has been closed.

Returns:

  • (Boolean) —

    true if #close has been called or the transport is closed



87
88
89
 
# File 'lib/kubemq/base_client.rb', line 87

def closed?
  @mutex.synchronize { @closed } || @transport.closed?
end

#create_channel(channel_name:, channel_type:) ⇒ Boolean

Creates a channel on the KubeMQ broker.

Parameters:

  • channel_name (String) —

    name for the new channel

  • channel_type (String) —

    one of ChannelType::EVENTS, EVENTS_STORE, COMMANDS, QUERIES, QUEUES

Returns:

  • (Boolean) —

    true on success

Raises:

See Also:



105
106
107
108
109
110
111
 
# File 'lib/kubemq/base_client.rb', line 105

def create_channel(channel_name:, channel_type:)
  ensure_not_closed!
  @transport.ensure_connected!
  Transport::ChannelManager.create_channel(
    @transport, @config.client_id, channel_name, channel_type
  )
end

#delete_channel(channel_name:, channel_type:) ⇒ Boolean

Deletes a channel from the KubeMQ broker.

Parameters:

  • channel_name (String) —

    name of the channel to delete

  • channel_type (String) —

    one of ChannelType constants

Returns:

  • (Boolean) —

    true on success

Raises:



123
124
125
126
127
128
129
 
# File 'lib/kubemq/base_client.rb', line 123

def delete_channel(channel_name:, channel_type:)
  ensure_not_closed!
  @transport.ensure_connected!
  Transport::ChannelManager.delete_channel(
    @transport, @config.client_id, channel_name, channel_type
  )
end

#list_channels(channel_type:, search: nil) ⇒ Array<ChannelInfo>

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

Examples:

channels = client.list_channels(channel_type: KubeMQ::ChannelType::EVENTS)
channels.each { |ch| puts "#{ch.name} active=#{ch.active?}" }

Parameters:

  • 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:



145
146
147
148
149
150
151
 
# File 'lib/kubemq/base_client.rb', line 145

def list_channels(channel_type:, search: nil)
  ensure_not_closed!
  @transport.ensure_connected!
  Transport::ChannelManager.list_channels(
    @transport, @config.client_id, channel_type, search
  )
end

#ping ⇒ ServerInfo

Pings the KubeMQ broker and returns server information.

Examples:

info = client.ping
puts "Connected to #{info.host} v#{info.version}"

Returns:

  • (ServerInfo) —

    broker host, version, and uptime

Raises:



67
68
69
70
71
 
# File 'lib/kubemq/base_client.rb', line 67

def ping
  ensure_not_closed!
  @transport.ensure_connected!
  @transport.ping
end

#purge_queue_channel(channel_name:) ⇒ Hash

Purges all messages from a queue channel.

Parameters:

  • channel_name (String) —

    queue channel to purge

Returns:

  • (Hash) —

    { affected_messages: Integer }

Raises:



161
162
163
164
165
166
167
 
# File 'lib/kubemq/base_client.rb', line 161

def purge_queue_channel(channel_name:)
  ensure_not_closed!
  @transport.ensure_connected!
  Transport::ChannelManager.purge_queue(
    @transport, @config.client_id, channel_name
  )
end