Class: Async::Redis::ClusterClient

Inherits:
Object
  • Object
show all
Includes:
Protocol::Redis::Cluster::Methods
Defined in:
lib/async/redis/cluster_client.rb

Overview

A Redis cluster client that manages multiple Redis instances and handles cluster operations.

Defined Under Namespace

Classes: Node, ReloadError, SlotError

Constant Summary collapse

HASH_SLOTS = 
16_384

Instance Method Summary collapse

Constructor Details

#initialize(endpoints, **options) ⇒ ClusterClient

Create a new instance of the cluster client.



33
34
35
36
37
 
# File 'lib/async/redis/cluster_client.rb', line 33

def initialize(endpoints, **options)
	@endpoints = endpoints
	@options = options
	@shards = nil
end

Instance Method Details

#any_client(role = :master) ⇒ Object

Get any available client from the cluster. This is useful for operations that don't require slot-specific routing, such as global pub/sub operations, INFO commands, or other cluster-wide operations.



92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
 
# File 'lib/async/redis/cluster_client.rb', line 92

def any_client(role = :master)
	unless @shards
		reload_cluster!
	end
	
	# Sample a random shard to get better load distribution
	if nodes = @shards.sample
		nodes = nodes.select{|node| node.role == role}
		
		if node = nodes.sample
			return (node.client ||= Client.new(node.endpoint, **@options))
		end
	end
	
	# Fallback to slot 0 if sampling fails
	client_for(0, role)
end

#client_for(slot, role = :master) ⇒ Object

Get a client for a specific slot.



72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
 
# File 'lib/async/redis/cluster_client.rb', line 72

def client_for(slot, role = :master)
	unless @shards
		reload_cluster!
	end
	
	if nodes = @shards.find(slot)
		nodes = nodes.select{|node| node.role == role}
	else
		raise SlotError, "No nodes found for slot #{slot}"
	end
	
	if node = nodes.sample
		return (node.client ||= Client.new(node.endpoint, **@options))
	end
end

#clients_for(*keys, role: :master, attempts: 3) ⇒ Object

Execute a block with clients for the given keys, grouped by cluster slot.



46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
 
# File 'lib/async/redis/cluster_client.rb', line 46

def clients_for(*keys, role: :master, attempts: 3)
	slots = slots_for(keys)
	
	slots.each do |slot, keys|
		yield client_for(slot, role), keys
	end
rescue ServerError => error
	Console.warn(self, error)
	
	if error.message =~ /MOVED|ASK/
		reload_cluster!
		
		attempts -= 1
		
		retry if attempts > 0
		
		raise
	else
		raise
	end
end

#slot_for(key) ⇒ Object

Return Redis::Client for a given key. Modified from https://github.com/antirez/redis-rb-cluster/blob/master/cluster.rb#L104-L117



201
202
203
204
205
206
207
208
209
210
211
 
# File 'lib/async/redis/cluster_client.rb', line 201

def slot_for(key)
	key = key.to_s
	
	if s = key.index("{")
		if e = key.index("}", s + 1) and e != s + 1
			key = key[s + 1..e - 1]
		end
	end
	
	return crc16(key) % HASH_SLOTS
end

#slots_for(keys) ⇒ Object

Calculate the hash slots for multiple keys.



216
217
218
219
220
221
222
223
224
 
# File 'lib/async/redis/cluster_client.rb', line 216

def slots_for(keys)
	slots = Hash.new{|hash, key| hash[key] = []}
	
	keys.each do |key|
		slots[slot_for(key)] << key
	end
	
	return slots
end

#subscribe(*channels) ⇒ Object

Subscribe to one or more sharded channels for pub/sub messaging in cluster environment. The subscription will be created on the appropriate nodes responsible for each channel's hash slot.



234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
 
# File 'lib/async/redis/cluster_client.rb', line 234

def subscribe(*channels)
	context = ClusterSubscription.new(self)
	
	if channels.any?
		context.subscribe(channels)
	end
	
	if block_given?
		begin
			yield context
		ensure
			context.close
		end
	else
		return context
	end
end