Class: Valkey::Glide::PubSub Private

Inherits:
Object
  • Object
show all
Defined in:
lib/valkey/glide/pubsub.rb

Overview

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

The wrapper around a Valkey client to handle all its PubSub functionalities.

Pub/Sub requires the RESP3 protocol, which parse_config enforces on the connect-time pubsub: option:

pubsub: {
subscriptions: {
  exact: ["news"],           # channel names
  pattern: ["news.*"],       # channel name patterns
  sharded: ["shard-chan"]    # sharded channels, cluster mode only
}
}

There are three ways to receive messages:

  • Inline: messages are queued and read with #get_message or #try_get_message.
  • Callback: a callback: proc receives every message instead, along with an arbitrary context:. Not supported yet.
  • Lazy: the _lazy subscribe and unsubscribe methods return without waiting for the server to confirm the change; read back #get_subscriptions for the subscriptions the server actually has.

Blocking methods take timeout: in seconds. glide-core receives it as an integer millisecond count appended as the last command argument, where 0 means "block indefinitely", so nil is sent as 0.

Only parse_config, the message queue and #close are implemented; the rest of the surface raises NotImplementedError.

See Also:

Defined Under Namespace

Modules: PushKind Classes: Message, SubscriptionState

Constant Summary collapse

SUBSCRIPTION_MODES =

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.

Subscription mode to the integer key glide-core expects in the connection JSON: :exact uses exact channel names, :pattern uses glob-style channel name patterns, :sharded uses sharded Pub/Sub and is cluster-only.

{ exact: 0, pattern: 1, sharded: 2 }.freeze
RESP3_VALUES =

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.

PubSub requires RESP3

[:resp3, "resp3", 3].freeze

Instance Attribute Summary collapse

Class Method Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(client, cluster_mode:) ⇒ PubSub

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.

Returns a new instance of PubSub.

Parameters:

  • client (Valkey) —

    A valkey client connection.

  • cluster_mode (Boolean) —

    The client cluster mode.



147
148
149
150
151
152
153
154
 
# File 'lib/valkey/glide/pubsub.rb', line 147

def initialize(client, cluster_mode:)
  @client = client
  @cluster_mode = cluster_mode
  @message_queue = Thread::Queue.new

  # The handler proc for receiving messages from the FFI.
  @ffi_handler = build_ffi_handler
end

Instance Attribute Details

#ffi_handler ⇒ Object (readonly)

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. 



156
157
158
 
# File 'lib/valkey/glide/pubsub.rb', line 156

def ffi_handler
  @ffi_handler
end

Class Method Details

.parse_config(pubsub_configs, protocol: nil) ⇒ Hash

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 and validates the pubsub: option.

Examples:

pubsub_configs:

{
  subscriptions: {
    exact:   ["news", "alerts"],  # exact matches
    pattern: ["news.*"],          # glob patterns
    sharded: ["shard-chan"]       # cluster mode
  },
  callback: ->(message, context) { ... },  # callback handler
  context: my_app_state                   # callback context
}

Parameters:

  • pubsub_configs (Hash, nil) —

    the pubsub: options: the channels and patterns to subscribe to at connection time under :subscriptions, keyed by mode, an optional :callback to accept the incoming messages and an arbitrary :context passed to it.

  • protocol (Symbol, String, Integer, nil) (defaults to: nil) —

    the protocol: option. Subscriptions require RESP3.

Returns:

  • (Hash) —

    the subscriptions keyed by the integer mode glide-core expects, or an empty Hash when nothing is subscribed to.

Raises:

  • (ArgumentError) —

    on an unknown subscription mode, or on subscriptions configured with a protocol other than RESP3.

See Also:



105
106
107
108
109
110
111
112
 
# File 'lib/valkey/glide/pubsub.rb', line 105

def parse_config(pubsub_configs, protocol: nil)
  pubsub_configs ||= {}
  subscriptions = pubsub_configs[:subscriptions] || {}

  validate!(subscriptions, protocol: protocol)

  to_ffi(pubsub_configs)
end

Instance Method Details

#close ⇒ 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.

Releases the Pub/Sub resources, closing the message queue. Callers blocked in #get_message are woken with nil.



178
179
180
 
# File 'lib/valkey/glide/pubsub.rb', line 178

def close
  @message_queue.close
end

#get_message ⇒ Message?

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.

Gets a Pub/Sub message, blocking until one is available.

Returns:

  • (Message, nil) —

    the next message, or nil once the message queue is closed.



162
163
164
 
# File 'lib/valkey/glide/pubsub.rb', line 162

def get_message
  @message_queue.pop
end

#get_subscriptions ⇒ SubscriptionState

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.

The connection's subscription state: what the client asked for, and what the server currently has. The way to confirm the outcome of a lazy subscribe or unsubscribe call.

Returns:

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.



394
 
# File 'lib/valkey/glide/pubsub.rb', line 394

def get_subscriptions = raise(NotImplementedError, "#{__method__} is not implemented yet")

#psubscribe(*patterns, timeout: 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.

Subscribes to channel patterns (blocking). Updates the client's desired subscription state and waits for the server's confirmation.

Parameters:

  • patterns (Array<String>) —

    the glob-style patterns to subscribe to, for example "news.*". An empty list is rejected with "No channels provided for subscription".

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



226
 
# File 'lib/valkey/glide/pubsub.rb', line 226

def psubscribe(*patterns, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#psubscribe_lazy(*patterns) ⇒ 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.

Subscribes to channel patterns (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. The client subscribes asynchronously in the background. Use #get_subscriptions to verify the actual server-side subscription state.

Parameters:

  • patterns (Array<String>) —

    the glob-style patterns to subscribe to, for example "news.*". An empty list is rejected with "No channels provided for subscription".

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



321
 
# File 'lib/valkey/glide/pubsub.rb', line 321

def psubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#publish(message, channel, sharded: false) ⇒ 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.

Publishes a message on a Pub/Sub channel. Aggregates the PUBLISH and SPUBLISH functionalities, the mode selected by sharded. In both modes the request is routed using the hashed channel as key. The message comes first, matching the other GLIDE clients.

Parameters:

  • message (String) —

    the message to publish.

  • channel (String) —

    the channel to publish the message on.

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

    use sharded Pub/Sub mode. Available since Valkey version 7.0, and requires cluster_mode: true.

Returns:

  • (Integer) —

    the number of subscriptions that received the message. In cluster mode that is the count on the node the request was routed to; in standalone it is the count on the primary node, which does not include subscriptions configured on replicas.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



386
 
# File 'lib/valkey/glide/pubsub.rb', line 386

def publish(message, channel, sharded: false) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub(subcommand, *args) ⇒ Array<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.

Raw PUBSUB passthrough, dispatching to the matching pubsub_* method.

Parameters:

  • subcommand (String, Symbol) —

    one of channels, numpat, numsub, shardchannels or shardnumsub.

  • args (Array) —

    the arguments of the subcommand.

Returns:

  • (Array<String>, Integer, Hash{String => Integer}) —

    whatever the target method returns.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



468
 
# File 'lib/valkey/glide/pubsub.rb', line 468

def pubsub(subcommand, *args) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub_channels(pattern = nil) ⇒ Array<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.

Lists the currently active channels, that is the channels with at least one exact subscriber. The command is routed to all nodes and the responses are aggregated into a single array.

Parameters:

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

    a glob-style pattern to match active channels against. nil returns all active channels.

Returns:

  • (Array<String>) —

    the active channels matching the given pattern.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



405
 
# File 'lib/valkey/glide/pubsub.rb', line 405

def pubsub_channels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub_numpat ⇒ 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.

Counts the unique patterns clients are subscribed to. That is the total number of unique patterns across all clients, not the number of clients subscribed to patterns. The command is routed to all nodes and the responses are aggregated into their sum.

Returns:

  • (Integer) —

    the number of unique patterns.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



415
 
# File 'lib/valkey/glide/pubsub.rb', line 415

def pubsub_numpat = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub_numsub(*channels) ⇒ Hash{String => 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.

Counts the subscribers of the given channels, exclusive of clients subscribed to patterns. The command is routed to all nodes and the responses are aggregated into a single Hash.

Parameters:

  • channels (Array<String>) —

    the channels to query for the number of subscribers. Calling this without channels is valid and returns an empty Hash.

Returns:

  • (Hash{String => Integer}) —

    the channel names mapped to their number of subscribers.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



428
 
# File 'lib/valkey/glide/pubsub.rb', line 428

def pubsub_numsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub_shardchannels(pattern = nil) ⇒ Array<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.

Lists the currently active sharded channels, that is the sharded channels with at least one subscriber. The command is routed to all nodes and the responses are aggregated into a single array.

Since: Valkey version 7.0.0.

Parameters:

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

    a glob-style pattern to match active sharded channels against. nil returns all active sharded channels.

Returns:

  • (Array<String>) —

    the active sharded channels matching the given pattern.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



442
 
# File 'lib/valkey/glide/pubsub.rb', line 442

def pubsub_shardchannels(pattern = nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#pubsub_shardnumsub(*channels) ⇒ Hash{String => 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.

Counts the subscribers of the given sharded channels, exclusive of clients subscribed to patterns. The command is routed to all nodes and the responses are aggregated into a single Hash.

Since: Valkey version 7.0.0.

Parameters:

  • channels (Array<String>) —

    the sharded channels to query for the number of subscribers. Calling this without channels is valid and returns an empty Hash.

Returns:

  • (Hash{String => Integer}) —

    the sharded channel names mapped to their number of subscribers.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



457
 
# File 'lib/valkey/glide/pubsub.rb', line 457

def pubsub_shardnumsub(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#punsubscribe(*patterns, timeout: 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.

Unsubscribes from channel patterns (blocking). Updates the client's desired subscription state and waits for the server's confirmation.

Parameters:

  • patterns (Array<String>) —

    the patterns to unsubscribe from. Empty unsubscribes from every pattern.

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



241
 
# File 'lib/valkey/glide/pubsub.rb', line 241

def punsubscribe(*patterns, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#punsubscribe_lazy(*patterns) ⇒ 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.

Unsubscribes from channel patterns (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. Use #get_subscriptions to verify the actual server-side subscription state.

Parameters:

  • patterns (Array<String>) —

    the patterns to unsubscribe from. Empty unsubscribes from every pattern.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



334
 
# File 'lib/valkey/glide/pubsub.rb', line 334

def punsubscribe_lazy(*patterns) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#ssubscribe(*channels, timeout: 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.

Subscribes to sharded channels (blocking). Updates the client's desired subscription state and waits for the server's confirmation. Requires cluster_mode: true; sharded Pub/Sub has no standalone equivalent.

Since: Valkey version 7.0.0.

Parameters:

  • channels (Array<String>) —

    the sharded channel names to subscribe to. An empty list is rejected with "No channels provided for subscription".

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



260
 
# File 'lib/valkey/glide/pubsub.rb', line 260

def ssubscribe(*channels, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#ssubscribe_lazy(*channels) ⇒ 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.

Subscribes to sharded channels (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. The client subscribes asynchronously in the background. Use #get_subscriptions to verify the actual server-side subscription state. Requires cluster_mode: true; sharded Pub/Sub has no standalone equivalent.

Since: Valkey version 7.0.0.

Parameters:

  • channels (Array<String>) —

    the sharded channel names to subscribe to. An empty list is rejected with "No channels provided for subscription".

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



352
 
# File 'lib/valkey/glide/pubsub.rb', line 352

def ssubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#subscribe(*channels, timeout: 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.

Subscribes to exact channels (blocking). Updates the client's desired subscription state and waits for the server's confirmation.

Parameters:

  • channels (Array<String>) —

    the channel names to subscribe to. An empty list is rejected with "No channels provided for subscription".

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



195
 
# File 'lib/valkey/glide/pubsub.rb', line 195

def subscribe(*channels, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#subscribe_lazy(*channels) ⇒ 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.

Subscribes to exact channels (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. The client subscribes asynchronously in the background. Use #get_subscriptions to verify the actual server-side subscription state.

Parameters:

  • channels (Array<String>) —

    the channel names to subscribe to. An empty list is rejected with "No channels provided for subscription".

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



293
 
# File 'lib/valkey/glide/pubsub.rb', line 293

def subscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#sunsubscribe(*channels, timeout: 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.

Unsubscribes from sharded channels (blocking). Updates the client's desired subscription state and waits for the server's confirmation. Requires cluster_mode: true; sharded Pub/Sub has no standalone equivalent.

Since: Valkey version 7.0.0.

Parameters:

  • channels (Array<String>) —

    the sharded channel names to unsubscribe from. Empty unsubscribes from every sharded channel.

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



279
 
# File 'lib/valkey/glide/pubsub.rb', line 279

def sunsubscribe(*channels, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#sunsubscribe_lazy(*channels) ⇒ 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.

Unsubscribes from sharded channels (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. Use #get_subscriptions to verify the actual server-side subscription state. Requires cluster_mode: true; sharded Pub/Sub has no standalone equivalent.

Since: Valkey version 7.0.0.

Parameters:

  • channels (Array<String>) —

    the sharded channel names to unsubscribe from. Empty unsubscribes from every sharded channel.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



368
 
# File 'lib/valkey/glide/pubsub.rb', line 368

def sunsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#try_get_message ⇒ Message?

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.

Gets a Pub/Sub message without blocking.

Returns:

  • (Message, nil) —

    the next message, or nil when the message queue is empty or closed.



170
171
172
173
174
 
# File 'lib/valkey/glide/pubsub.rb', line 170

def try_get_message
  @message_queue.pop(true)
rescue ThreadError
  nil
end

#unsubscribe(*channels, timeout: 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.

Unsubscribes from exact channels (blocking). Updates the client's desired subscription state and waits for the server's confirmation.

Parameters:

  • channels (Array<String>) —

    the channel names to unsubscribe from. Empty unsubscribes from every exact channel.

  • timeout (Float, Integer, nil) (defaults to: nil) —

    maximum time in seconds to wait for the server's confirmation. nil blocks indefinitely.

Raises:

  • (ArgumentError) —

    on a negative timeout.

  • (Valkey::TimeoutError) —

    when the timeout expires before the server confirms.

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



210
 
# File 'lib/valkey/glide/pubsub.rb', line 210

def unsubscribe(*channels, timeout: nil) = raise(NotImplementedError, "#{__method__} is not implemented yet")

#unsubscribe_lazy(*channels) ⇒ 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.

Unsubscribes from exact channels (non-blocking). Updates the client's desired subscription state without waiting for the server's confirmation, and returns as soon as the local state is updated. Use #get_subscriptions to verify the actual server-side subscription state.

Parameters:

  • channels (Array<String>) —

    the channel names to unsubscribe from. Empty unsubscribes from every exact channel.

Raises:

  • (NotImplementedError) —

    this method is not implemented yet.

See Also:



306
 
# File 'lib/valkey/glide/pubsub.rb', line 306

def unsubscribe_lazy(*channels) = raise(NotImplementedError, "#{__method__} is not implemented yet")