flutter/shorebird
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

refactor(redis_client): extract RedisConnection from RedisClient (#3796)

Browse Source
This commit is contained in:
nickshorebird
2026-05-22 13:23:01 -04:00
committed by GitHub
parent d42474dc8a
commit 57a6193efc
7 changed files with 242 additions and 155 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.github/workflows/_shorebird_ci_dart.yaml
+9
View File
@@ -21,6 +21,10 @@ on:
required: false
default: ""
type: string
needs_redis:
required: false
default: false
type: boolean
jobs:
ci:
@@ -29,6 +33,11 @@ jobs:
- uses: actions/checkout@v6
with:
submodules: recursive
- name: 🐳 Run Redis
if: inputs.needs_redis
run: |
docker pull redis/redis-stack-server:latest
docker run --name test_redis -d -p 6379:6379 --rm -e REDIS_ARGS="--requirepass password" redis/redis-stack-server:latest
- uses: dart-lang/setup-dart@v1
- name: Setup Bloc Tools
if: inputs.has_bloc_lint
.github/workflows/shorebird_ci.yaml
+1
View File
@@ -204,6 +204,7 @@ jobs:
has_bloc_lint: false
has_unit_tests: true
subpackages: ""
needs_redis: true
stripe_api:
needs: changes
packages/redis_client/CHANGELOG.md
+4
View File
@@ -1,3 +1,7 @@
# 0.0.13
- refactor: extract `RedisConnection` from `RedisClient` to prepare for opt-in connection pooling
# 0.0.12
- fix: more redis query result type handling fixes
packages/redis_client/lib/src/redis_client.dart
+20 -154
View File
@@ -6,6 +6,8 @@ import 'package:resp_client/resp_client.dart';
import 'package:resp_client/resp_commands.dart';
import 'package:resp_client/resp_server.dart';
part 'redis_connection.dart';
/// {@template redis_exception}
/// An exception thrown by the Redis client.
/// {@endtemplate}
@@ -116,43 +118,18 @@ class RedisClient {
RedisSocketOptions socket = const RedisSocketOptions(),
RedisCommandOptions command = const RedisCommandOptions(),
RedisLogger logger = const _NoopRedisLogger(),
}) : _socketOptions = socket,
_commandOptions = command,
_logger = logger;
/// The socket options for the Redis server.
final RedisSocketOptions _socketOptions;
}) : _commandOptions = command,
_logger = logger,
_connection = RedisConnection(options: socket, logger: logger);
/// The command options for the Redis client.
final RedisCommandOptions _commandOptions;
/// The underlying connection to the Redis server.
RespServerConnection? _connection;
/// The logger for the Redis client.
final RedisLogger _logger;
/// The underlying client for interacting with the Redis server.
RespClient? _client;
/// Whether the client has been closed.
var _closed = false;
/// A completer which completes when the client establishes a connection.
var _connected = Completer<void>();
/// Whether the client is connected.
var _isConnected = false;
/// A completer which completes when the client disconnects.
/// Begins in a completed state since the client is initially disconnected.
var _disconnected = Completer<void>()..complete();
/// A future which completes when the client establishes a connection.
Future<void> get _untilConnected => _connected.future;
/// A future which completes when the client disconnects.
Future<void> get _untilDisconnected => _disconnected.future;
/// The underlying connection to the Redis server.
final RedisConnection _connection;
/// The Redis JSON commands.
RedisJson get json => RedisJson._(client: this);
@@ -265,137 +242,34 @@ class RedisClient {
}
/// Send a command to the Redis server.
Future<dynamic> execute(List<Object?> command) async {
return _runWithRetry(() async {
final result = await RespCommandsTier0(_client!).execute(command);
if (result.isError) throw RedisException(result.toString());
return result.payload;
}, command: command.join(' '));
Future<dynamic> execute(List<Object?> command) {
return _runWithRetry(
() => _connection.execute(command),
command: command.join(' '),
);
}
/// Establish a connection to the Redis server.
/// The delay between connection attempts.
Future<void> connect() async {
if (_closed) throw StateError('RedisClient has been closed.');
unawaited(_reconnect(retryAttempts: _socketOptions.retryAttempts));
return _untilConnected;
}
Future<void> connect() => _connection.connect();
/// Terminate the connection to the Redis server.
Future<void> disconnect() async {
_logger.info('Disconnecting.');
await _connection?.close();
_reset();
await _untilDisconnected;
_logger.info('Disconnected.');
}
Future<void> disconnect() => _connection.disconnect();
/// Terminate the connection to the Redis server and close the client.
/// After this method is called, the client instance is no longer usable.
/// Call this method when you are done using the client and/or wish to
/// prevent reconnection attempts.
Future<void> close() {
_logger.info('Closing connection.');
_closed = true;
return disconnect();
}
Future<void> _reconnect({required int retryAttempts}) async {
if (retryAttempts <= 0) {
_connected.completeError(
const SocketException('Connection retry limit exceeded'),
StackTrace.current,
);
return;
}
Future<void> onConnectionOpened(RespServerConnection connection) async {
_logger.info('Connection opened.');
_disconnected = Completer<void>();
_connection = connection;
_client = RespClient(connection);
if (_socketOptions.password != null) {
_logger.info('Authenticating.');
final username = _socketOptions.username;
final password = _socketOptions.password!;
await RespCommandsTier0(_client!).execute(['AUTH', username, password]);
}
_isConnected = true;
if (!_connected.isCompleted) _connected.complete();
_logger.info('Connected.');
}
void onConnectionClosed([Object? error, StackTrace? stackTrace]) {
if (error == null) {
_logger.info('Connection closed.');
} else {
_logger.error(
'Connection closed with error.',
error: error,
stackTrace: stackTrace,
);
}
if (_closed) return;
final wasConnected = _isConnected;
_isConnected = false;
final retryInterval = _socketOptions.retryInterval;
final totalAttempts = _socketOptions.retryAttempts;
final remainingAttempts = wasConnected
? totalAttempts
: retryAttempts - 1;
final attemptsMade = totalAttempts - remainingAttempts;
final attemptInfo = attemptsMade > 0
? ' ($attemptsMade/$totalAttempts attempts)'
: '';
if (wasConnected) _reset();
_logger.info(
'Reconnecting in ${retryInterval.inMilliseconds}ms$attemptInfo.',
);
Future<void>.delayed(
retryInterval,
() => _reconnect(retryAttempts: remainingAttempts),
);
}
try {
_logger.info('Connecting to ${_socketOptions.connectionUri}.');
final uri = _socketOptions.connectionUri;
final connection = await connectSocket(
uri.host,
port: uri.port,
timeout: _socketOptions.timeout,
);
unawaited(onConnectionOpened(connection));
unawaited(
connection.outputSink.done
.then((_) => onConnectionClosed())
.catchError(onConnectionClosed),
);
} on Exception catch (error, stackTrace) {
onConnectionClosed(error, stackTrace);
}
}
void _reset() {
_connected = Completer<void>();
_connection = null;
_client = null;
if (!_disconnected.isCompleted) _disconnected.complete();
}
Future<void> close() => _connection.close();
Future<T> _runWithRetry<T>(
Future<T> Function() fn, {
required String command,
int? remainingAttempts,
}) async {
if (_closed) throw StateError('RedisClient has been closed.');
if (_connection.isClosed) {
throw StateError('RedisClient has been closed.');
}
final totalAttempts = _commandOptions.retryAttempts;
remainingAttempts ??= _commandOptions.retryAttempts;
@@ -407,10 +281,7 @@ class RedisClient {
_logger.debug('Executing "$command"$attemptInfo.');
try {
return await Future<T>.sync(() async {
await _untilConnected;
return fn();
}).timeout(_commandOptions.timeout);
return await Future<T>.sync(fn).timeout(_commandOptions.timeout);
} catch (error, stackTrace) {
if (error is RedisException) rethrow;
if (remainingAttempts > 0) {
@@ -431,7 +302,7 @@ class RedisClient {
error: error,
stackTrace: stackTrace,
);
await _connection?.close();
await _connection.recycle();
rethrow;
}
}
@@ -988,11 +859,6 @@ class RedisTDigest {
}
}
extension on RedisSocketOptions {
/// The connection URI for the Redis server derived from the socket options.
Uri get connectionUri => Uri.parse('redis://$host:$port');
}
final class _NoopRedisLogger implements RedisLogger {
const _NoopRedisLogger();
packages/redis_client/lib/src/redis_connection.dart
+193
View File
@@ -0,0 +1,193 @@
part of 'redis_client.dart';
/// {@template redis_connection}
/// A single transport-level connection to a Redis server.
///
/// Owns the socket lifecycle: opening, authenticating, reconnect-on-drop,
/// and graceful close. Commands sent through [execute] wait for the socket
/// to be ready before being written to the wire.
///
/// This type is internal to the package today and is not exported. A future
/// release will expose it so callers can pin a connection for transactions
/// and pipelining.
/// {@endtemplate}
class RedisConnection {
/// {@macro redis_connection}
RedisConnection({required this.options, required this.logger});
/// The socket-level options used to dial and re-dial the server.
final RedisSocketOptions options;
/// The logger used for connection-lifecycle events.
final RedisLogger logger;
/// The underlying connection to the Redis server.
RespServerConnection? _connection;
/// The underlying RESP client for the active connection.
RespClient? _client;
/// Whether the connection has been permanently closed.
var _closed = false;
/// A completer which completes when the socket is established.
var _connected = Completer<void>();
/// Whether the socket is currently established.
var _isConnected = false;
/// A completer which completes when the socket is torn down.
/// Begins in a completed state since the connection is initially down.
var _disconnected = Completer<void>()..complete();
/// A future which completes when the socket is established.
Future<void> get _untilConnected => _connected.future;
/// A future which completes when the socket is torn down.
Future<void> get _untilDisconnected => _disconnected.future;
/// Whether [close] has been called. After this returns true the connection
/// is no longer usable.
bool get isClosed => _closed;
/// Open the socket. Returns once the connection is established.
Future<void> connect() async {
if (_closed) throw StateError('RedisClient has been closed.');
unawaited(_reconnect(retryAttempts: options.retryAttempts));
return _untilConnected;
}
/// Tear down the current socket without closing the connection logically.
/// The reconnect loop will recreate the socket on subsequent use unless
/// [close] has been called.
Future<void> disconnect() async {
logger.info('Disconnecting.');
await _connection?.close();
_reset();
await _untilDisconnected;
logger.info('Disconnected.');
}
/// Permanently close the connection. After this method is called, the
/// instance is not usable.
Future<void> close() {
logger.info('Closing connection.');
_closed = true;
return disconnect();
}
/// Send a single command to the server. Waits for the socket to be ready
/// before writing. Throws [RedisException] on a server-side error reply.
/// Throws [StateError] if the connection has been closed.
Future<dynamic> execute(List<Object?> command) async {
if (_closed) throw StateError('RedisClient has been closed.');
await _untilConnected;
final result = await RespCommandsTier0(_client!).execute(command);
if (result.isError) throw RedisException(result.toString());
return result.payload;
}
/// Close the underlying socket without closing the connection logically.
/// The reconnect loop will recreate the socket on next [execute].
///
/// Used by [RedisClient]'s retry policy to recover from a wedged socket
/// after the retry budget is exhausted.
Future<void> recycle() async {
await _connection?.close();
}
Future<void> _reconnect({required int retryAttempts}) async {
if (retryAttempts <= 0) {
_connected.completeError(
const SocketException('Connection retry limit exceeded'),
StackTrace.current,
);
return;
}
Future<void> onConnectionOpened(RespServerConnection connection) async {
logger.info('Connection opened.');
_disconnected = Completer<void>();
_connection = connection;
_client = RespClient(connection);
if (options.password != null) {
logger.info('Authenticating.');
final username = options.username;
final password = options.password!;
await RespCommandsTier0(_client!).execute(['AUTH', username, password]);
}
_isConnected = true;
if (!_connected.isCompleted) _connected.complete();
logger.info('Connected.');
}
void onConnectionClosed([Object? error, StackTrace? stackTrace]) {
if (error == null) {
logger.info('Connection closed.');
} else {
logger.error(
'Connection closed with error.',
error: error,
stackTrace: stackTrace,
);
}
if (_closed) return;
final wasConnected = _isConnected;
_isConnected = false;
final retryInterval = options.retryInterval;
final totalAttempts = options.retryAttempts;
final remainingAttempts = wasConnected
? totalAttempts
: retryAttempts - 1;
final attemptsMade = totalAttempts - remainingAttempts;
final attemptInfo = attemptsMade > 0
? ' ($attemptsMade/$totalAttempts attempts)'
: '';
if (wasConnected) _reset();
logger.info(
'Reconnecting in ${retryInterval.inMilliseconds}ms$attemptInfo.',
);
Future<void>.delayed(
retryInterval,
() => _reconnect(retryAttempts: remainingAttempts),
);
}
try {
logger.info('Connecting to ${options._connectionUri}.');
final uri = options._connectionUri;
final connection = await connectSocket(
uri.host,
port: uri.port,
timeout: options.timeout,
);
unawaited(onConnectionOpened(connection));
unawaited(
connection.outputSink.done
.then((_) => onConnectionClosed())
.catchError(onConnectionClosed),
);
} on Exception catch (error, stackTrace) {
onConnectionClosed(error, stackTrace);
}
}
void _reset() {
_connected = Completer<void>();
_connection = null;
_client = null;
if (!_disconnected.isCompleted) _disconnected.complete();
}
}
extension on RedisSocketOptions {
/// The connection URI for the Redis server derived from the socket options.
Uri get _connectionUri => Uri.parse('redis://$host:$port');
}
packages/redis_client/pubspec.yaml
+1 -1
View File
@@ -1,6 +1,6 @@
name: shorebird_redis_client
description: A lightweight Dart client library for communicating with a Redis server. Built by Shorebird.
version: 0.0.12
version: 0.0.13
homepage: https://shorebird.dev
repository: https://github.com/shorebirdtech/shorebird/tree/main/packages/redis_client
topics: [redis, cache, shorebird]
packages/redis_client/test/src/redis_client_test.dart
+14
View File
@@ -108,6 +108,20 @@ void main() {
),
);
});
test('execute throws StateError when closed', () async {
await client.close();
await expectLater(
client.execute(['PING']),
throwsA(
isA<StateError>().having(
(e) => e.message,
'message',
'RedisClient has been closed.',
),
),
);
});
});
group('disconnect', () {