From b422593d77bbd7b29bf97021fead0070f821af89 Mon Sep 17 00:00:00 2001 From: Felix Angelov Date: Fri, 15 Sep 2023 11:32:27 -0500 Subject: [PATCH] feat(redis_client): add logging support (#1270) --- packages/redis_client/example/main.dart | 23 +- packages/redis_client/lib/redis_client.dart | 8 +- .../redis_client/lib/src/redis_client.dart | 290 +++++++++++++----- .../test/src/redis_client_test.dart | 129 ++++++-- 4 files changed, 313 insertions(+), 137 deletions(-) diff --git a/packages/redis_client/example/main.dart b/packages/redis_client/example/main.dart index d5c3d288..80abd69a 100644 --- a/packages/redis_client/example/main.dart +++ b/packages/redis_client/example/main.dart @@ -1,5 +1,3 @@ -import 'dart:async'; - import 'package:redis_client/redis_client.dart'; Future main() async { @@ -9,24 +7,9 @@ Future main() async { // Connect to the Redis server. await client.connect(); - const key = 'HELLO'; + // Execute a command. + await client.execute(['PING']); // PONG - final initialValue = await client.get(key: key); // null - assert(initialValue == null, 'Key should not exist.'); - - // Set the value of a key. - await client.set(key: key, value: 'WORLD'); - - // Get the value of a key. - final value = await client.get(key: key); // WORLD - assert(value == 'WORLD', 'Value should be "WORLD".'); - - // Delete the key. - await client.delete(key: key); - - final finalValue = await client.get(key: key); // null - assert(finalValue == null, 'Key should not exist.'); - - // Close the connection to the Redis server. + // Close the connection. await client.close(); } diff --git a/packages/redis_client/lib/redis_client.dart b/packages/redis_client/lib/redis_client.dart index cda72108..97543190 100644 --- a/packages/redis_client/lib/redis_client.dart +++ b/packages/redis_client/lib/redis_client.dart @@ -1,2 +1,8 @@ export 'src/redis_client.dart' - show RedisClient, RedisCommandOptions, RedisJson, RedisSocketOptions; + show + RedisClient, + RedisCommandOptions, + RedisException, + RedisJson, + RedisLogger, + RedisSocketOptions; diff --git a/packages/redis_client/lib/src/redis_client.dart b/packages/redis_client/lib/src/redis_client.dart index b997a060..d2f42a48 100644 --- a/packages/redis_client/lib/src/redis_client.dart +++ b/packages/redis_client/lib/src/redis_client.dart @@ -6,6 +6,20 @@ import 'package:resp_client/resp_client.dart'; import 'package:resp_client/resp_commands.dart'; import 'package:resp_client/resp_server.dart'; +/// {@template redis_exception} +/// An exception thrown by the Redis client. +/// {@endtemplate} +class RedisException implements Exception { + /// {@macro redis_exception} + const RedisException(this.message); + + /// The message for the exception. + final String message; + + @override + String toString() => message; +} + /// {@template redis_socket_options} /// Options for connecting to a Redis server. /// {@endtemplate} @@ -14,7 +28,11 @@ class RedisSocketOptions { const RedisSocketOptions({ this.host = 'localhost', this.port = 6379, + this.username = 'default', + this.password, this.timeout = const Duration(seconds: 30), + this.retryInterval = const Duration(seconds: 1), + this.retryAttempts = 10, }); /// The host of the Redis server. @@ -28,6 +46,22 @@ class RedisSocketOptions { /// The timeout for connecting to the Redis server. /// Defaults to 30 seconds. final Duration timeout; + + /// The username for authenticating to the Redis server. + /// Defaults to 'default'. + final String username; + + /// The password for authenticating to the Redis server. + /// Defaults to null. + final String? password; + + /// The delay between connection attempts. + /// Defaults to 1 second. + final Duration retryInterval; + + /// The maximum number of connection attempts. + /// Defaults to 10. + final int retryAttempts; } /// {@template redis_command_options} @@ -37,11 +71,40 @@ class RedisCommandOptions { /// {@macro redis_command_options} const RedisCommandOptions({ this.timeout = const Duration(seconds: 10), + this.retryInterval = const Duration(seconds: 1), + this.retryAttempts = 3, }); /// The timeout for sending commands to the Redis server. /// Defaults to 10 seconds. final Duration timeout; + + /// The delay between command attempts. + /// Defaults to 1 second. + final Duration retryInterval; + + /// The maximum number of command attempts. + /// Defaults to 3. + final int retryAttempts; +} + +/// {@template redis_logger} +/// A logger for the Redis client. +/// {@endtemplate} +abstract interface class RedisLogger { + // coverage:ignore-start + /// {@macro redis_logger} + const RedisLogger(); + // coverage:ignore-end + + /// Log a debug message. + void debug(String message); + + /// Log an info message. + void info(String message); + + /// Log an error message. + void error(String message, {Object? error, StackTrace? stackTrace}); } /// {@template redis_client} @@ -52,8 +115,10 @@ class RedisClient { RedisClient({ RedisSocketOptions socket = const RedisSocketOptions(), RedisCommandOptions command = const RedisCommandOptions(), + RedisLogger logger = const _NoopRedisLogger(), }) : _socketOptions = socket, - _commandOptions = command; + _commandOptions = command, + _logger = logger; /// The socket options for the Redis server. final RedisSocketOptions _socketOptions; @@ -64,6 +129,9 @@ class RedisClient { /// 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; @@ -73,84 +141,81 @@ class RedisClient { /// A completer which completes when the client establishes a connection. var _connected = Completer(); + /// 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()..complete(); /// A future which completes when the client establishes a connection. - Future get connected => _connected.future; + Future get _untilConnected => _connected.future; /// A future which completes when the client disconnects. - Future get disconnected => _disconnected.future; + Future get _untilDisconnected => _disconnected.future; /// The Redis JSON commands. RedisJson get json => RedisJson._(client: this); /// Authenticate to the Redis server. - /// Returns true if successful, otherwise false. /// Equivalent to the `AUTH` command. /// https://redis.io/commands/auth - Future auth({ + Future auth({ required String password, String username = 'default', - }) async { - final result = await _exec( - () => sendCommand(['AUTH', username, password]), - ); - if (result is RespSimpleString) return result.payload == 'OK'; - return false; + }) { + return execute(['AUTH', username, password]); } /// Set the value of a key. /// Equivalent to the `SET` command. /// https://redis.io/commands/set Future set({required String key, required String value}) { - return _exec(() => RespCommandsTier2(_client!).set(key, value)); + return execute(['SET', key, value]); } /// Gets the value of a key. /// Returns null if the key does not exist. /// Equivalent to the `GET` command. /// https://redis.io/commands/get - Future get({required String key}) { - return _exec(() => RespCommandsTier2(_client!).get(key)); + Future get({required String key}) async { + return await execute(['GET', key]) as String?; } /// Deletes the specified key. /// Equivalent to the `DEL` command. /// https://redis.io/commands/del - Future delete({required String key}) { - return _exec(() => RespCommandsTier2(_client!).del([key])); - } + Future delete({required String key}) => execute(['DEL', key]); /// Send a command to the Redis server. - Future> sendCommand(List command) async { - return _exec(() => RespCommandsTier0(_client!).execute(command)); + Future execute(List 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(' '), + ); } /// Establish a connection to the Redis server. /// The delay between connection attempts. - Future connect({ - Duration connectionRetryDelay = const Duration(milliseconds: 100), - int maxConnectionAttempts = 100, - }) async { + Future connect() async { if (_closed) throw StateError('RedisClient has been closed.'); - unawaited( - _reconnect( - connectionRetryDelay: connectionRetryDelay, - remainingConnectionAttempts: maxConnectionAttempts, - ), - ); + unawaited(_reconnect(retryAttempts: _socketOptions.retryAttempts)); - return connected; + return _untilConnected; } /// Terminate the connection to the Redis server. - Future disconnect() { - _connection?.close(); + Future disconnect() async { + _logger.info('Disconnecting.'); + await _connection?.close(); _reset(); - return disconnected; + await _untilDisconnected; + _logger.info('Disconnected.'); } /// Terminate the connection to the Redis server and close the client. @@ -158,68 +223,87 @@ class RedisClient { /// Call this method when you are done using the client and/or wish to /// prevent reconnection attempts. Future close() { + _logger.info('Closing connection.'); _closed = true; return disconnect(); } - Future _reconnect({ - required Duration connectionRetryDelay, - required int remainingConnectionAttempts, - Object? error, - StackTrace? stackTrace, - }) async { - if (_closed) return; - - if (remainingConnectionAttempts <= 0) { + Future _reconnect({required int retryAttempts}) async { + if (retryAttempts <= 0) { _connected.completeError( - error ?? const SocketException('Connection retry limit exceeded'), - stackTrace, + const SocketException('Connection retry limit exceeded'), + StackTrace.current, ); return; } - void onConnectionOpened(RespServerConnection connection) { + Future onConnectionOpened(RespServerConnection connection) async { + _logger.info('Connection opened.'); _disconnected = Completer(); _connection = connection; _client = RespClient(connection); - _connected.complete(); + 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]) { - _reset(); - _reconnect( - connectionRetryDelay: connectionRetryDelay, - remainingConnectionAttempts: remainingConnectionAttempts - 1, - error: 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.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, ); - - onConnectionOpened(connection); - + unawaited(onConnectionOpened(connection)); unawaited( connection.outputSink.done .then((_) => onConnectionClosed()) .catchError(onConnectionClosed), ); } catch (error, stackTrace) { - Future.delayed( - connectionRetryDelay, - () => _reconnect( - connectionRetryDelay: connectionRetryDelay, - remainingConnectionAttempts: remainingConnectionAttempts - 1, - error: error, - stackTrace: stackTrace, - ), - ); + onConnectionClosed(error, stackTrace); } } @@ -230,16 +314,49 @@ class RedisClient { if (!_disconnected.isCompleted) _disconnected.complete(); } - Future _exec(FutureOr Function() fn) async { + Future _runWithRetry( + Future Function() fn, { + required String command, + int? remainingAttempts, + }) async { if (_closed) throw StateError('RedisClient has been closed.'); - await connected; - return Future.sync(fn).timeout( - _commandOptions.timeout, - onTimeout: () { - _connection?.close(); - throw const SocketException('Connection timed out'); - }, - ); + + final totalAttempts = _commandOptions.retryAttempts; + remainingAttempts ??= _commandOptions.retryAttempts; + final attemptsMade = totalAttempts - remainingAttempts; + final attemptInfo = + attemptsMade > 0 ? ' ($attemptsMade/$totalAttempts attempts)' : ''; + + _logger.debug('Executing "$command"$attemptInfo.'); + + try { + return await Future.sync(() async { + await _untilConnected; + return fn(); + }).timeout(_commandOptions.timeout); + } catch (error, stackTrace) { + if (error is RedisException) rethrow; + if (remainingAttempts > 0) { + _logger.error( + 'Command failed to complete. Retrying.', + error: error, + stackTrace: stackTrace, + ); + return _runWithRetry( + fn, + command: command, + remainingAttempts: remainingAttempts - 1, + ); + } + + _logger.error( + 'Command failed to complete.', + error: error, + stackTrace: stackTrace, + ); + await _connection?.close(); + rethrow; + } } } @@ -260,7 +377,7 @@ class RedisJson { required String key, required Map value, }) { - return _client.sendCommand(['JSON.SET', key, r'$', json.encode(value)]); + return _client.execute(['JSON.SET', key, r'$', json.encode(value)]); } /// Gets the value of a key. @@ -268,13 +385,9 @@ class RedisJson { /// Equivalent to the `JSON.GET` command. /// https://redis.io/commands/json.get Future?> get({required String key}) async { - final result = await _client.sendCommand([ - 'JSON.GET', - key, - r'$', - ]); - if (result is RespBulkString) { - final parts = LineSplitter.split(result.payload ?? ''); + final result = await _client.execute(['JSON.GET', key, r'$']); + if (result is String) { + final parts = LineSplitter.split(result); if (parts.isNotEmpty) { final decoded = json.decode(parts.first) as List; if (decoded.isNotEmpty) return decoded.first as Map; @@ -287,7 +400,7 @@ class RedisJson { /// Equivalent to the `JSON.DEL` command. /// https://redis.io/commands/json.del Future delete({required String key}) { - return _client.sendCommand(['JSON.DEL', key, r'$']); + return _client.execute(['JSON.DEL', key, r'$']); } } @@ -295,3 +408,16 @@ 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(); + + @override + void debug(String message) {} + + @override + void info(String message) {} + + @override + void error(String message, {Object? error, StackTrace? stackTrace}) {} +} diff --git a/packages/redis_client/test/src/redis_client_test.dart b/packages/redis_client/test/src/redis_client_test.dart index 22275194..7311cf91 100644 --- a/packages/redis_client/test/src/redis_client_test.dart +++ b/packages/redis_client/test/src/redis_client_test.dart @@ -1,3 +1,4 @@ +import 'dart:async'; import 'dart:io'; import 'package:redis_client/redis_client.dart'; @@ -8,59 +9,74 @@ void main() { late RedisClient client; setUp(() async { - client = RedisClient(); - await client.connect(); + client = RedisClient( + socket: const RedisSocketOptions(password: 'password'), + ); }); tearDown(() async { - try { - await client.sendCommand(['RESET']); - await client.sendCommand(['FLUSHALL']); - await client.close(); - } catch (_) { - // ignore - } + await client.close(); + }); + + group(RedisException, () { + test('overrides toString', () { + const message = 'A RedisException occurred.'; + const exception = RedisException(message); + expect(exception.toString(), equals(message)); + }); }); group('connect', () { - test('throws SocketException when connection times out', () async { + test('authenticates automatically when credentials are provided', + () async { + await expectLater(client.connect(), completes); + await expectLater(client.execute(['PING']), completion(equals('PONG'))); + }); + + test('throws SocketException when connection times out w/retry', + () async { final client = RedisClient( - socket: const RedisSocketOptions(timeout: Duration(microseconds: 1)), + socket: const RedisSocketOptions( + timeout: Duration(microseconds: 1), + retryAttempts: 1, + ), ); await expectLater( - () => client.connect(maxConnectionAttempts: 1), + client.connect, throwsA( isA().having( (e) => e.message, 'message', - contains('Connection timed out'), + contains('Connection retry limit exceeded'), ), ), ); + await client.close(); }); test('throws SocketException after max connection attempts', () async { final client = RedisClient( - socket: const RedisSocketOptions(port: 1234), + socket: const RedisSocketOptions(port: 1234, retryAttempts: 1), ); await expectLater( - () => client.connect(maxConnectionAttempts: 1), + client.connect, throwsA( isA().having( (e) => e.message, 'message', - contains('Connection refused'), + contains('Connection retry limit exceeded'), ), ), ); + await client.close(); }); - test('throws SocketException after disconnect', () async { + test('throws SocketException after disconnect w/out retry', () async { final client = RedisClient( - socket: const RedisSocketOptions(port: 1234), + socket: const RedisSocketOptions(port: 1234, retryAttempts: 0), ); await expectLater( - () => client.connect(maxConnectionAttempts: 0), + client.connect, throwsA( isA().having( (e) => '$e', @@ -69,6 +85,7 @@ void main() { ), ), ); + await client.close(); }); test('throws StateError when closed', () async { @@ -86,47 +103,86 @@ void main() { }); }); + group('disconnect', () { + test('closes the connection and reconnects', () async { + await client.connect(); + await client.disconnect(); + await expectLater(client.execute(['PING']), completion(equals('PONG'))); + }); + }); + group('AUTH', () { + setUp(() async { + await client.connect(); + }); + test('is required', () async { + final client = RedisClient(); + await client.connect(); await expectLater( client.get(key: 'foo'), throwsA( - isA().having( + isA().having( (e) => e.message, 'message', contains('-NOAUTH Authentication required.'), ), ), ); + await client.close(); }); test('fails when username is incorrect', () async { await expectLater( client.auth(username: 'shorebird', password: 'password'), - completion(isFalse), + throwsA( + isA().having( + (e) => e.message, + 'message', + contains('-WRONGPASS invalid username-password pair'), + ), + ), ); }); test('fails when password is incorrect', () async { await expectLater( client.auth(password: 'oops'), - completion(isFalse), + throwsA( + isA().having( + (e) => e.message, + 'message', + contains('-WRONGPASS invalid username-password pair'), + ), + ), ); }); test('succeeds when username/password are correct', () async { await expectLater( client.auth(password: 'password'), - completion(isTrue), + completes, ); }); }); group('GET/SET/DEL', () { + setUp(() async { + await client.connect(); + }); + + tearDown(() async { + try { + await client.execute(['RESET']); + await client.execute(['FLUSHALL']); + } catch (_) { + // ignore + } + }); + test('completes', () async { const key = 'key'; const value = 'value'; - await client.auth(password: 'password'); await expectLater(client.get(key: key), completion(isNull)); await expectLater(client.set(key: key, value: value), completes); await expectLater(client.get(key: key), completion(equals(value))); @@ -135,27 +191,33 @@ void main() { }); test( - 'throws SocketException ' + 'throws TimeoutException ' 'when command timeout is exceeded', () async { final client = RedisClient( command: const RedisCommandOptions(timeout: Duration.zero), ); - await client.connect(); await expectLater( client.get(key: 'foo'), - throwsA( - isA().having( - (e) => e.message, - 'message', - contains('Connection timed out'), - ), - ), + throwsA(isA()), ); }); }); group('JSON', () { group('GET/SET/DEL', () { + setUp(() async { + await client.connect(); + }); + + tearDown(() async { + try { + await client.execute(['RESET']); + await client.execute(['FLUSHALL']); + } catch (_) { + // ignore + } + }); + test('completes', () async { const key = 'key'; const value = { @@ -164,7 +226,6 @@ void main() { 'nested': {'bar': 42}, 'array': [1, 2, 3], }; - await client.auth(password: 'password'); await expectLater(client.json.get(key: key), completion(isNull)); await expectLater(client.json.set(key: key, value: value), completes); await expectLater(