Class: Async::Redis::ClusterSubscription

Inherits:
Object
  • Object
show all
Defined in:
lib/async/redis/cluster_subscription.rb

Overview

Context for managing sharded subscriptions across multiple Redis cluster nodes. This class handles the complexity of subscribing to channels that may be distributed across different shards in a Redis cluster.

Defined Under Namespace

Classes: SubscriptionError

Instance Method Summary collapse

Constructor Details

#initialize(cluster_client, queue: Async::LimitedQueue.new) ⇒ ClusterSubscription

Initialize a new shard subscription context.



21
22
23
24
25
26
27
28
 
# File 'lib/async/redis/cluster_subscription.rb', line 21

def initialize(cluster_client, queue: Async::LimitedQueue.new)
	@cluster_client = cluster_client
	@subscriptions = {}
	@channels = []
	
	@barrier = Async::Barrier.new
	@queue = queue
end

Instance Method Details

#channels ⇒ Object

Get the list of currently subscribed channels.



118
119
120
 
# File 'lib/async/redis/cluster_subscription.rb', line 118

def channels
	@channels.dup
end

#close ⇒ Object

Close all shard subscriptions.



31
32
33
34
35
36
37
38
39
 
# File 'lib/async/redis/cluster_subscription.rb', line 31

def close
	if barrier = @barrier
		@barrier = nil
		barrier.stop
	end
	
	@subscriptions.each_value(&:close)
	@subscriptions.clear
end

#each ⇒ Object

Iterate over all messages from all subscribed shards.



53
54
55
56
57
58
59
 
# File 'lib/async/redis/cluster_subscription.rb', line 53

def each
	return to_enum unless block_given?
	
	while response = self.listen
		yield response
	end
end

#listen ⇒ Object

Listen for the next message from any subscribed shard.



44
45
46
47
48
 
# File 'lib/async/redis/cluster_subscription.rb', line 44

def listen
	@queue.pop
rescue => error
	raise SubscriptionError, "Failed to read message!"
end

#shard_count ⇒ Object

Get the number of active shard subscriptions.



124
125
126
 
# File 'lib/async/redis/cluster_subscription.rb', line 124

def shard_count
	@subscriptions.size
end

#subscribe(channels) ⇒ Object

Subscribe to additional sharded channels.



63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
 
# File 'lib/async/redis/cluster_subscription.rb', line 63

def subscribe(channels)
	slots = @cluster_client.slots_for(channels)
	
	slots.each do |slot, channels_for_slot|
		if subscription = @subscriptions[slot]
			# Add to existing subscription for this shard
			subscription.ssubscribe(channels_for_slot)
		else
			# Create new subscription for this shard
			client = @cluster_client.client_for(slot)
			subscription = @subscriptions[slot] = client.ssubscribe(*channels_for_slot)
			
			@barrier.async do
				# This is optimistic, in other words, subscription.listen will also fail on close.
				until subscription.closed?
					@queue << subscription.listen
				end
			ensure
				# If we are exiting here for any reason OTHER than the subscription was closed, we need to re-create the subscription state:
				unless subscription.closed?
					@queue.close
				end
			end
		end
	end
	
	@channels.concat(channels)
end

#unsubscribe(channels) ⇒ Object

Unsubscribe from sharded channels.



94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
 
# File 'lib/async/redis/cluster_subscription.rb', line 94

def unsubscribe(channels)
	slots = @cluster_client.slots_for(channels)
	
	slots.each do |slot, channels_for_slot|
		if subscription = @subscriptions[slot]
			subscription.sunsubscribe(channels_for_slot)
			
			# Remove channels from our tracking
			@channels -= channels_for_slot
			
			# Check if this shard still has channels
			remaining_channels_for_slot = @channels.select{|ch| @cluster_client.slot_for(ch) == slot}
			
			# If no channels left for this shard, close and remove it
			if remaining_channels_for_slot.empty?
				@subscriptions.delete(slot)
				subscription.close
			end
		end
	end
end