Class: SolidRedis::ClusterClient

Inherits:
Object
  • Object
show all
Defined in:
lib/solid_redis/cluster_client.rb

Overview

Routes commands to the cluster node owning their key slot and follows MOVED/ASK redirections. Holds one Client per node; all of them belong to the Ractor that created this object.

Constant Summary collapse

REDIRECTION = 
/\A(MOVED|ASK) (\d+) (\S+):(\d+)\z/
RETRY_DELAY = 
0.05
TRANSIENT_ERRORS = 
%w[TRYAGAIN CLUSTERDOWN].freeze

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, name: nil) ⇒ ClusterClient

Returns a new instance of ClusterClient.



14
15
16
17
18
 
# File 'lib/solid_redis/cluster_client.rb', line 14

def initialize(config, name: nil)
  @config = config
  @name = name
  @clients = {}
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



12
13
14
 
# File 'lib/solid_redis/cluster_client.rb', line 12

def config
  @config
end

Instance Method Details

#blocking_call(timeout, *command) ⇒ Object 



28
29
30
 
# File 'lib/solid_redis/cluster_client.rb', line 28

def blocking_call(timeout, *command)
  blocking_call_v(timeout, command)
end

#blocking_call_v(timeout, command) ⇒ Object 



32
33
34
 
# File 'lib/solid_redis/cluster_client.rb', line 32

def blocking_call_v(timeout, command)
  route(command, retry_connection: false) { |client| client.blocking_call_v(timeout, command) }
end

#call(*command) ⇒ Object 



20
21
22
 
# File 'lib/solid_redis/cluster_client.rb', line 20

def call(*command)
  call_v(command)
end

#call_v(command) ⇒ Object 



24
25
26
 
# File 'lib/solid_redis/cluster_client.rb', line 24

def call_v(command)
  route(command) { |client| client.call_v(command) }
end

#close ⇒ Object 



74
75
76
77
78
79
 
# File 'lib/solid_redis/cluster_client.rb', line 74

def close
  clients = @clients.values
  @clients = {}
  clients.each(&:close)
  self
end

#connected? ⇒ Boolean

Returns:

  • (Boolean) 


70
71
72
 
# File 'lib/solid_redis/cluster_client.rb', line 70

def connected?
  @clients.values.any?(&:connected?)
end

#pipelined(exception: true) {|pipeline| ... } ⇒ Object

Groups commands by node, runs one pipeline per node and restores the original order. Redirected commands are replayed individually.

Yields:

  • (pipeline) 


38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
 
# File 'lib/solid_redis/cluster_client.rb', line 38

def pipelined(exception: true)
  pipeline = Client::Pipeline.new
  yield pipeline
  commands = pipeline.commands
  return [] if commands.empty?

  results = Array.new(commands.length)
  commands.each_with_index.group_by { |command, _| node_for(command) }.each do |node, group|
    replies = client_for(node).pipelined(exception: false) do |batch|
      group.each { |command, _| batch.call_v(command) }
    end
    group.each_with_index { |(_, index), position| results[index] = replies[position] }
  end

  results.each_with_index do |result, index|
    next unless result.is_a?(CommandError)
    next unless result.message.match?(REDIRECTION) || transient_error?(result)

    results[index] = begin
      call_v(commands[index])
    rescue CommandError => error
      error
    end
  end

  if exception && (error = results.find { |result| result.is_a?(CommandError) })
    raise error
  end

  results
end

#server_url ⇒ Object 



81
82
83
 
# File 'lib/solid_redis/cluster_client.rb', line 81

def server_url
  @clients.keys.join(",")
end