Class: RedisClient::Cluster::Router

Inherits:
Object
  • Object
show all
Defined in:
lib/redis_client/cluster/router.rb

Defined Under Namespace

Classes: Action

Instance Attribute Summary collapse

Instance Method Summary collapse

Constructor Details

#initialize(config, concurrent_worker, pool: nil, **kwargs) ⇒ Router

Returns a new instance of Router.



32
33
34
35
36
37
38
39
40
41
42
43
44
45
 
# File 'lib/redis_client/cluster/router.rb', line 32

def initialize(config, concurrent_worker, pool: nil, **kwargs)
  @config = config
  @concurrent_worker = concurrent_worker
  @pool = pool
  @client_kwargs = kwargs
  @node = ::RedisClient::Cluster::Node.new(concurrent_worker, config: config, pool: pool, **kwargs)
  @node.reload!
  @command = ::RedisClient::Cluster::Command.load(@node.replica_clients.shuffle, slow_command_timeout: config.slow_command_timeout)
  @command_builder = @config.command_builder
  @dedicated_actions = build_dedicated_actions
rescue ::RedisClient::Cluster::InitialSetupError => e
  e.with_config(config)
  raise
end

Instance Attribute Details

#config ⇒ Object (readonly)

Returns the value of attribute config.



23
24
25
 
# File 'lib/redis_client/cluster/router.rb', line 23

def config
  @config
end

Instance Method Details

#assign_asking_node(err_msg) ⇒ Object 



239
240
241
242
 
# File 'lib/redis_client/cluster/router.rb', line 239

def assign_asking_node(err_msg)
  _, _, node_key = err_msg.split
  handle_node_reload_error { @node.find_by(node_key) }
end

#assign_node(command) ⇒ Object 



174
175
176
177
178
179
 
# File 'lib/redis_client/cluster/router.rb', line 174

def assign_node(command)
  handle_node_reload_error do
    node_key = find_node_key(command)
    @node.find_by(node_key)
  end
end

#assign_node_and_send_command(method, command, args, retry_count: 3, &block) ⇒ Object

See Also:



80
81
82
83
 
# File 'lib/redis_client/cluster/router.rb', line 80

def assign_node_and_send_command(method, command, args, retry_count: 3, &block)
  node = assign_node(command)
  send_command_to_node(node, method, command, args, retry_count: retry_count, &block)
end

#assign_redirection_node(err_msg) ⇒ Object 



232
233
234
235
236
237
 
# File 'lib/redis_client/cluster/router.rb', line 232

def assign_redirection_node(err_msg)
  _, slot, node_key = err_msg.split
  slot = slot.to_i
  @node.update_slot(slot, node_key)
  handle_node_reload_error { @node.find_by(node_key) }
end

#close ⇒ Object 



254
255
256
 
# File 'lib/redis_client/cluster/router.rb', line 254

def close
  @node.each(&:close)
end

#command_exists?(name) ⇒ Boolean

Returns:

  • (Boolean) 


228
229
230
 
# File 'lib/redis_client/cluster/router.rb', line 228

def command_exists?(name)
  @command.exists?(name)
end

#find_node(node_key) ⇒ Object 



224
225
226
 
# File 'lib/redis_client/cluster/router.rb', line 224

def find_node(node_key)
  handle_node_reload_error { @node.find_by(node_key) }
end

#find_node_key(command, seed: nil) ⇒ Object 



202
203
204
205
 
# File 'lib/redis_client/cluster/router.rb', line 202

def find_node_key(command, seed: nil)
  key = @command.extract_first_key(command)
  find_node_key_by_key(key, seed: seed, primary: @command.should_send_to_primary?(command))
end

#find_node_key_by_key(key, seed: nil, primary: false) ⇒ Object 



181
182
183
184
185
186
187
188
189
190
191
192
193
 
# File 'lib/redis_client/cluster/router.rb', line 181

def find_node_key_by_key(key, seed: nil, primary: false)
  if key && !key.empty?
    slot = ::RedisClient::Cluster::KeySlotConverter.convert(key)
    node_key = primary ? @node.find_node_key_of_primary(slot) : @node.find_node_key_of_replica(slot)
    if node_key.nil?
      renew_cluster_state
      raise ::RedisClient::Cluster::NodeMightBeDown.new.with_config(@config)
    end
    node_key
  else
    primary ? @node.any_primary_node_key(seed: seed) : @node.any_replica_node_key(seed: seed)
  end
end

#find_primary_node_by_slot(slot) ⇒ Object 



195
196
197
198
199
200
 
# File 'lib/redis_client/cluster/router.rb', line 195

def find_primary_node_by_slot(slot)
  handle_node_reload_error do
    node_key = @node.find_node_key_of_primary(slot)
    @node.find_by(node_key)
  end
end

#find_primary_node_key(command) ⇒ Object 



207
208
209
210
211
212
 
# File 'lib/redis_client/cluster/router.rb', line 207

def find_primary_node_key(command)
  key = @command.extract_first_key(command)
  return nil unless key&.size&.> 0

  find_node_key_by_key(key, primary: true)
end

#find_slot(command) ⇒ Object 



214
215
216
 
# File 'lib/redis_client/cluster/router.rb', line 214

def find_slot(command)
  find_slot_by_key(@command.extract_first_key(command))
end

#find_slot_by_key(key) ⇒ Object 



218
219
220
221
222
 
# File 'lib/redis_client/cluster/router.rb', line 218

def find_slot_by_key(key)
  return if key.empty?

  ::RedisClient::Cluster::KeySlotConverter.convert(key)
end

#handle_redirection(node, command, retry_count:) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



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
128
129
130
131
132
133
134
135
136
137
 
# File 'lib/redis_client/cluster/router.rb', line 96

def handle_redirection(node, command, retry_count:) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  yield node
rescue ::RedisClient::CircuitBreaker::OpenCircuitError
  raise
rescue ::RedisClient::CommandError => e
  raise unless ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(e, node)

  retry_count -= 1
  if e.message.start_with?('MOVED')
    node = assign_redirection_node(e.message)
    retry if retry_count >= 0
  elsif e.message.start_with?('ASK')
    node = assign_asking_node(e.message)
    if retry_count >= 0
      node.call('asking')
      retry
    end
  elsif e.message.start_with?('CLUSTERDOWN')
    renew_cluster_state
    retry if retry_count >= 0
  end

  raise
rescue ::RedisClient::ConnectionError => e
  raise unless ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(e, node)

  retry_count -= 1
  renew_cluster_state

  if retry_count >= 0
    # Find the node to use for this command - if this fails for some reason, though, re-use
    # the old node.
    begin
      node = find_node(find_node_key(command)) if command
    rescue StandardError # rubocop:disable Lint/SuppressedException
    end
    retry
  end

  retry if retry_count >= 0
  raise
end

#node_keys ⇒ Object 



244
245
246
 
# File 'lib/redis_client/cluster/router.rb', line 244

def node_keys
  @node.node_keys
end

#renew_cluster_state ⇒ Object 



248
249
250
251
252
 
# File 'lib/redis_client/cluster/router.rb', line 248

def renew_cluster_state
  @node.reload!
rescue ::RedisClient::Cluster::InitialSetupError
  # ignore
end

#scan(command, seed: nil) ⇒ Object

rubocop:disable Metrics/AbcSize



139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
 
# File 'lib/redis_client/cluster/router.rb', line 139

def scan(command, seed: nil) # rubocop:disable Metrics/AbcSize
  command[1] = ZERO_CURSOR_FOR_SCAN if command.size == 1
  input_cursor = Integer(command[1])

  client_index = input_cursor % 256
  raw_cursor = input_cursor >> 8

  clients = @node.clients_for_scanning(seed: seed)

  client = clients[client_index]
  return [ZERO_CURSOR_FOR_SCAN, []] unless client

  command[1] = raw_cursor.to_s

  result_cursor, result_keys = client.call_v(command)
  result_cursor = Integer(result_cursor)

  client_index += 1 if result_cursor == 0

  [((result_cursor << 8) + client_index).to_s, result_keys]
rescue ::RedisClient::ConnectionError
  renew_cluster_state
  raise
end

#scan_single_key(command, arity:, &block) ⇒ Object 



164
165
166
167
168
169
170
171
172
 
# File 'lib/redis_client/cluster/router.rb', line 164

def scan_single_key(command, arity:, &block)
  node = assign_node(command)
  loop do
    cursor, values = handle_redirection(node, nil, retry_count: 3) { |n| n.call_v(command) }
    command[2] = cursor
    arity < 2 ? values.each(&block) : values.each_slice(arity, &block)
    break if cursor == ZERO_CURSOR_FOR_SCAN
  end
end

#send_command(method, command, *args, &block) ⇒ Object

rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity



47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
 
# File 'lib/redis_client/cluster/router.rb', line 47

def send_command(method, command, *args, &block) # rubocop:disable Metrics/AbcSize, Metrics/CyclomaticComplexity, Metrics/PerceivedComplexity
  return assign_node_and_send_command(method, command, args, &block) unless @dedicated_actions.key?(command.first)

  action = @dedicated_actions[command.first]
  return send(action.method_name, method, command, args, &block) if action.reply_transformer.nil?

  reply = send(action.method_name, method, command, args)
  action.reply_transformer.call(reply).then(&TSF.call(block))
rescue ::RedisClient::CircuitBreaker::OpenCircuitError
  raise
rescue ::RedisClient::Cluster::Node::ReloadNeeded
  renew_cluster_state
  raise ::RedisClient::Cluster::NodeMightBeDown.new.with_config(@config)
rescue ::RedisClient::ConnectionError
  renew_cluster_state
  raise
rescue ::RedisClient::CommandError => e
  renew_cluster_state if e.message.start_with?('CLUSTERDOWN')
  raise
rescue ::RedisClient::Cluster::ErrorCollection => e
  e.with_config(@config)
  raise if e.errors.any?(::RedisClient::CircuitBreaker::OpenCircuitError)

  renew_cluster_state if e.errors.values.any? do |err|
    next false if ::RedisClient::Cluster::ErrorIdentification.identifiable?(err) && @node.none? { |c| ::RedisClient::Cluster::ErrorIdentification.client_owns_error?(err, c) }

    err.message.start_with?('CLUSTERDOWN') || err.is_a?(::RedisClient::ConnectionError)
  end

  raise
end

#send_command_to_node(node, method, command, args, retry_count: 3, &block) ⇒ Object 



85
86
87
88
89
90
91
92
93
94
 
# File 'lib/redis_client/cluster/router.rb', line 85

def send_command_to_node(node, method, command, args, retry_count: 3, &block)
  handle_redirection(node, command, retry_count: retry_count) do |on_node|
    if args.empty?
      # prevent memory allocation for variable-length args
      on_node.public_send(method, command, &block)
    else
      on_node.public_send(method, *args, command, &block)
    end
  end
end