Class: Async::Redis::Context::Subscription

Inherits:
Generic
  • Object
show all
Defined in:
lib/async/redis/context/subscription.rb

Overview

Context for Redis pub/sub subscription operations.

Constant Summary collapse

MESSAGE = 
"message"
PMESSAGE = 
"pmessage"
SMESSAGE = 
"smessage"

Instance Method Summary collapse

Methods inherited from Generic

#call, #closed?, #read_response, #write_request

Constructor Details

#initialize(pool, channels) ⇒ Subscription

Initialize a new subscription context.



21
22
23
24
25
 
# File 'lib/async/redis/context/subscription.rb', line 21

def initialize(pool, channels)
	super(pool)
	
	subscribe(channels) if channels.any?
end

Instance Method Details

#close ⇒ Object

Close the subscription context.



28
29
30
31
32
33
 
# File 'lib/async/redis/context/subscription.rb', line 28

def close
	# This causes anyone calling `#listen` to exit, as `read_response` will fail. If we decided to use `RESET` instead, we'd need to take that into account.
	@connection&.close
	
	super
end

#each ⇒ Object

Iterate over all messages from subscribed channels.



50
51
52
53
54
55
56
 
# File 'lib/async/redis/context/subscription.rb', line 50

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

#listen ⇒ Object

Listen for the next message from subscribed channels.



37
38
39
40
41
42
43
44
45
 
# File 'lib/async/redis/context/subscription.rb', line 37

def listen
	while response = @connection.read_response
		type = response.first
		
		if type == MESSAGE || type == PMESSAGE || type == SMESSAGE
			return response
		end
	end
end

#psubscribe(patterns) ⇒ Object

Subscribe to channel patterns.



74
75
76
77
 
# File 'lib/async/redis/context/subscription.rb', line 74

def psubscribe(patterns)
	@connection.write_request ["PSUBSCRIBE", *patterns]
	@connection.flush
end

#punsubscribe(patterns) ⇒ Object

Unsubscribe from channel patterns.



81
82
83
84
 
# File 'lib/async/redis/context/subscription.rb', line 81

def punsubscribe(patterns)
	@connection.write_request ["PUNSUBSCRIBE", *patterns]
	@connection.flush
end

#ssubscribe(channels) ⇒ Object

Subscribe to sharded channels (Redis 7.0+).



88
89
90
91
 
# File 'lib/async/redis/context/subscription.rb', line 88

def ssubscribe(channels)
	@connection.write_request ["SSUBSCRIBE", *channels]
	@connection.flush
end

#subscribe(channels) ⇒ Object

Subscribe to additional channels.



60
61
62
63
 
# File 'lib/async/redis/context/subscription.rb', line 60

def subscribe(channels)
	@connection.write_request ["SUBSCRIBE", *channels]
	@connection.flush
end

#sunsubscribe(channels) ⇒ Object

Unsubscribe from sharded channels (Redis 7.0+).



95
96
97
98
 
# File 'lib/async/redis/context/subscription.rb', line 95

def sunsubscribe(channels)
	@connection.write_request ["SUNSUBSCRIBE", *channels]
	@connection.flush
end

#unsubscribe(channels) ⇒ Object

Unsubscribe from channels.



67
68
69
70
 
# File 'lib/async/redis/context/subscription.rb', line 67

def unsubscribe(channels)
	@connection.write_request ["UNSUBSCRIBE", *channels]
	@connection.flush
end