Initial rework of sentinels client.

Author: ioquatix

.github/workflows/sentinel.conf

@@ -0,0 +1,5 @@
+port 26379
+sentinel monitor mymaster redis-master 6379 1
+sentinel down-after-milliseconds mymaster 1000
+sentinel failover-timeout mymaster 1000
+sentinel parallel-syncs mymaster 1

.github/workflows/test-sentinel.yaml

@@ -0,0 +1,84 @@
+name: Test
+
+on: [push, pull_request]
+
+permissions:
+  contents: read
+
+env:
+  CONSOLE_OUTPUT: XTerm
+  ASYNC_REDIS_MASTER: "redis-master:6379"
+  ASYNC_REDIS_SLAVE: "redis-slave:6379"
+  ASYNC_REDIS_SENTINEL: "redis-sentinel:26379"
+
+jobs:
+  test:
+    name: ${{matrix.ruby}} on ${{matrix.os}}
+    runs-on: ${{matrix.os}}-latest
+    continue-on-error: ${{matrix.experimental}}
+    
+    services:
+      redis-master:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 6379
+      redis-slave:
+        image: redis
+        options: >-
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 6379
+      redis-sentinel:
+        image: redis
+        volumes:
+          - ./sentinel.conf:/etc/redis/sentinel.conf
+        options: >-
+          --entrypoint "redis-sentinel /etc/redis/sentinel.conf"
+          --health-cmd "redis-cli ping"
+          --health-interval 10s
+          --health-timeout 5s
+          --health-retries 5
+        ports:
+          - 26379
+    
+    strategy:
+      matrix:
+        os:
+          - ubuntu
+        
+        ruby:
+          - "3.1"
+          - "3.2"
+          - "3.3"
+        
+        experimental: [false]
+        
+        include:
+          - os: ubuntu
+            ruby: truffleruby
+            experimental: true
+          - os: ubuntu
+            ruby: jruby
+            experimental: true
+          - os: ubuntu
+            ruby: head
+            experimental: true
+    
+    steps:
+    - uses: actions/checkout@v4
+    - uses: ruby/setup-ruby@v1
+      with:
+        ruby-version: ${{matrix.ruby}}
+        bundler-cache: true
+    
+    - name: Run tests
+      timeout-minutes: 10
+      run: bundle exec bake test

lib/async/redis.rb

@@ -6,4 +6,4 @@
 
 require_relative 'redis/version'
 require_relative 'redis/client'
-require_relative 'redis/sentinels'
+require_relative 'redis/sentinel_client'

lib/async/redis/client.rb

@@ -26,6 +26,65 @@ module Redis
 		class Client
 			include ::Protocol::Redis::Methods
 
+			module Methods
+				def subscribe(*channels)
+					context = Context::Subscribe.new(@pool, channels)
+					
+					return context unless block_given?
+					
+					begin
+						yield context
+					ensure
+						context.close
+					end
+				end
+				
+				def transaction(&block)
+					context = Context::Transaction.new(@pool)
+					
+					return context unless block_given?
+					
+					begin
+						yield context
+					ensure
+						context.close
+					end
+				end
+				
+				alias multi transaction
+				
+				def pipeline(&block)
+					context = Context::Pipeline.new(@pool)
+					
+					return context unless block_given?
+					
+					begin
+						yield context
+					ensure
+						context.close
+					end
+				end
+				
+				# Deprecated.
+				alias nested pipeline
+				
+				def call(*arguments)
+					@pool.acquire do |connection|
+						connection.write_request(arguments)
+						
+						connection.flush
+						
+						return connection.read_response
+					end
+				end
+				
+				def close
+					@pool.close
+				end
+			end
+			
+			include Methods
+			
 			def initialize(endpoint = Endpoint.local, protocol: endpoint.protocol, **options)
 				@endpoint = endpoint
 				@protocol = protocol
@@ -38,8 +97,8 @@ def initialize(endpoint = Endpoint.local, protocol: endpoint.protocol, **options
 
 			# @return [client] if no block provided.
 			# @yield [client, task] yield the client in an async task.
-			def self.open(*arguments, &block)
-				client = self.new(*arguments)
+			def self.open(*arguments, **options, &block)
+				client = self.new(*arguments, **options)
 
 				return client unless block_given?
 
@@ -52,61 +111,6 @@ def self.open(*arguments, &block)
 				end.wait
 			end
 
-			def close
-				@pool.close
-			end
-			
-			def subscribe(*channels)
-				context = Context::Subscribe.new(@pool, channels)
-				
-				return context unless block_given?
-				
-				begin
-					yield context
-				ensure
-					context.close
-				end
-			end
-			
-			def transaction(&block)
-				context = Context::Transaction.new(@pool)
-				
-				return context unless block_given?
-				
-				begin
-					yield context
-				ensure
-					context.close
-				end
-			end
-			
-			alias multi transaction
-			
-			def pipeline(&block)
-				context = Context::Pipeline.new(@pool)
-				
-				return context unless block_given?
-				
-				begin
-					yield context
-				ensure
-					context.close
-				end
-			end
-			
-			# Deprecated.
-			alias nested pipeline
-			
-			def call(*arguments)
-				@pool.acquire do |connection|
-					connection.write_request(arguments)
-					
-					connection.flush
-					
-					return connection.read_response
-				end
-			end
-			
 			protected
 
 			def connect(**options)

lib/async/redis/endpoint.rb

@@ -23,7 +23,11 @@ class Endpoint < ::IO::Endpoint::Generic
 
 			def self.local(**options)
 				self.new(LOCALHOST, **options)
-			end	
+			end
+			
+			def self.remote(host, port = 6379, **options)
+				self.new(URI.parse("redis://#{host}:#{port}"), **options)
+			end
 
 			SCHEMES = {
 				'redis' => URI::Generic,

lib/async/redis/sentinels.rb renamed to lib/async/redis/sentinel_client.rb

@@ -4,27 +4,29 @@
 # Copyright, 2020, by David Ortiz.
 # Copyright, 2023-2024, by Samuel Williams.
 
+require_relative 'client'
 require 'io/stream'
 
 module Async
 	module Redis
-		class SentinelsClient < Client
-			def initialize(master_name, sentinels, role = :master, protocol = Protocol::RESP2, **options)
+		class SentinelClient
+			DEFAULT_MASTER_NAME = 'mymaster'
+			
+			include ::Protocol::Redis::Methods
+			include Client::Methods
+			
+			def initialize(endpoints, master_name: DEFAULT_MASTER_NAME, role: :master, protocol: Protocol::RESP2, **options)
+				@endpoints = endpoints
 				@master_name = master_name
-				
-				@sentinel_endpoints = sentinels.map do |sentinel|
-					::IO::Endpoint.tcp(sentinel[:host], sentinel[:port])
-				end
-				
 				@role = role
 				@protocol = protocol
+				
 				@pool = connect(**options)
 			end
 
-			private
+			protected
 
-			# Override the parent method. The only difference is that this one needs
-			# to resolve the master/slave address.
+			# Override the parent method. The only difference is that this one needs to resolve the master/slave address.
 			def connect(**options)
 				Async::Pool::Controller.wrap(**options) do
 					endpoint = resolve_address
@@ -49,24 +51,24 @@ def resolve_address
 			end
 
 			def resolve_master
-				@sentinel_endpoints.each do |sentinel_endpoint|
-					client = Client.new(sentinel_endpoint, protocol: @protocol)
+				@endpoints.each do |endpoint|
+					client = Client.new(endpoint)
 
 					begin
 						address = client.call('sentinel', 'get-master-addr-by-name', @master_name)
 					rescue Errno::ECONNREFUSED
 						next
 					end
 
-					return ::IO::Endpoint.tcp(address[0], address[1]) if address
+					return Endpoint.remote(address[0], address[1]) if address
 				end
 
-				nil
+				return nil
 			end
 
 			def resolve_slave
-				@sentinel_endpoints.each do |sentinel_endpoint|
-					client = Client.new(sentinel_endpoint, protocol: @protocol)
+				@endpoints.each do |endpoint|
+					client = Client.new(endpoint)
 
 					begin
 						reply = client.call('sentinel', 'slaves', @master_name)
@@ -78,10 +80,10 @@ def resolve_slave
 					next if slaves.empty?
 
 					slave = select_slave(slaves)
-					return ::IO::Endpoint.tcp(slave['ip'], slave['port'])
+					return Endpoint.remote(slave['ip'], slave['port'])
 				end
 
-				nil
+				return nil
 			end
 
 			def available_slaves(reply)

test/async/redis/sentinel_client.rb

@@ -0,0 +1,24 @@
+# frozen_string_literal: true
+
+# Released under the MIT License.
+# Copyright, 2018-2024, by Samuel Williams.
+# Copyright, 2018, by Huba Nagy.
+# Copyright, 2019, by David Ortiz.
+
+require 'async/clock'
+require 'async/redis/sentinel_client'
+require 'sus/fixtures/async'
+
+describe Async::Redis::SentinelClient do
+	include Sus::Fixtures::Async::ReactorContext
+	
+	let(:master_host) {ENV['ASYNC_REDIS_MASTER']}
+	let(:slave_host) {ENV['ASYNC_REDIS_SLAVE']}
+	let(:sentinel_host) {ENV['ASYNC_REDIS_SENTINEL']}
+	
+	it "should resolve master address" do
+		unless master_host and slave_host and sentinel_host
+			skip("No sentinel host provided.")
+		end
+	end
+end