feat(redis_client): basic RedisClient (#1266)

.github/workflows/main.yaml

@@ -18,11 +18,9 @@ jobs:
   changes:
     runs-on: ubuntu-latest
 
-    permissions:
-      pull-requests: read
-
     outputs:
       needs_dart_build: ${{ steps.needs_dart_build.outputs.changes }}
+      needs_redis_build: ${{ steps.needs_redis_build.outputs.changes }}
       needs_verify: ${{ steps.needs_verify.outputs.changes }}
 
     name: 👀 Detect Changes
@@ -48,10 +46,6 @@ jobs:
               - ./.github/workflows/main.yaml
               - ./.github/actions/dart_package/action.yaml
               - packages/discord_gcp_alerts/**
-            redis_client:
-              - ./.github/workflows/main.yaml
-              - ./.github/actions/dart_package/action.yaml
-              - packages/redis_client/**
             shorebird_cli:
               - ./.github/workflows/main.yaml
               - ./.github/actions/dart_package/action.yaml
@@ -76,6 +70,16 @@ jobs:
               - ./.github/actions/dart_package/action.yaml
               - packages/scoped/**
 
+      - uses: dorny/paths-filter@v2
+        name: Redis Detection
+        id: needs_redis_build
+        with:
+          filters: |
+            redis_client:
+              - ./.github/workflows/main.yaml
+              - ./.github/actions/dart_package/action.yaml
+              - packages/redis_client/**
+
       - uses: dorny/paths-filter@v2
         name: Verify Detection
         id: needs_verify
@@ -113,6 +117,35 @@ jobs:
           working_directory: packages/${{ matrix.package }}
           min_coverage: ${{ matrix.package == 'cutler' && '10' || '100' }}
 
+  build_redis:
+    needs: changes
+    if: ${{ needs.changes.outputs.needs_redis_build != '[]' }}
+
+    permissions: write-all
+
+    strategy:
+      matrix:
+        package: ${{ fromJSON(needs.changes.outputs.needs_redis_build) }}
+
+    runs-on: ubuntu-latest
+
+    name: 🎯 Build ${{ matrix.package }}
+
+    steps:
+      - name: 📚 Git Checkout
+        uses: actions/checkout@v3
+
+      - name: 🐳 Run Redis
+        run: |
+          docker pull redis:latest
+          docker run --name test_redis -d -p 6379:6379 redis redis-server --requirepass "password"
+
+      - name: 🎯 Build ${{ matrix.package }}
+        uses: ./.github/actions/dart_package
+        with:
+          codecov_token: ${{ secrets.CODECOV_TOKEN }}
+          working_directory: packages/${{ matrix.package }}
+
   verify_packages:
     needs: changes
     if: ${{ needs.changes.outputs.needs_verify != '[]' }}
@@ -135,7 +168,8 @@ jobs:
           working_directory: packages/${{ matrix.package }}
 
   ci:
-    needs: [semantic_pull_request, build_dart_packages, verify_packages]
+    needs:
+      [semantic_pull_request, build_dart_packages, build_redis, verify_packages]
     if: ${{ always() }}
 
     runs-on: ubuntu-latest

packages/redis_client/README.md

@@ -7,6 +7,34 @@ A Dart library for interacting with [Redis](https://redis.io).
 [license_badge]: https://img.shields.io/badge/license-MIT-blue.svg
 [license_link]: https://opensource.org/licenses/MIT
 
+## Quick Start
+
+```dart
+import 'dart:async';
+
+import 'package:redis_client/redis_client.dart';
+
+Future<void> main() async {
+  // Create an instance of a RedisClient.
+  final client = RedisClient();
+
+  // Connect to the Redis server.
+  await client.connect();
+
+  // Set the value of a key.
+  await client.set(key: 'HELLO', value: 'WORLD');
+
+  // Get the value of a key.
+  final value = await client.get(key: 'HELLO'); // WORLD
+
+  // Delete the key.
+  await client.delete(key: 'HELLO');
+
+  // Close the connection to the Redis server.
+  await client.close();
+}
+```
+
 ## License
 
 Shorebird packages are licensed for use under either Apache License, Version 2.0

packages/redis_client/analysis_options.yaml

@@ -1 +1 @@
-include: package:very_good_analysis/analysis_options.5.0.0.yaml
+include: package:very_good_analysis/analysis_options.5.1.0.yaml

packages/redis_client/example/main.dart

@@ -0,0 +1,32 @@
+import 'dart:async';
+
+import 'package:redis_client/redis_client.dart';
+
+Future<void> main() async {
+  // Create an instance of a RedisClient.
+  final client = RedisClient();
+
+  // Connect to the Redis server.
+  await client.connect();
+
+  const key = 'HELLO';
+
+  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.
+  await client.close();
+}

packages/redis_client/lib/redis_client.dart

@@ -1 +1,2 @@
-export 'src/redis_client.dart' show RedisClient;
+export 'src/redis_client.dart'
+    show RedisClient, RedisCommandOptions, RedisSocketOptions;

packages/redis_client/lib/src/redis_client.dart

@@ -1,7 +1,245 @@
+import 'dart:async';
+import 'dart:io';
+
+import 'package:resp_client/resp_client.dart';
+import 'package:resp_client/resp_commands.dart';
+import 'package:resp_client/resp_server.dart';
+
+/// {@template redis_socket_options}
+/// Options for connecting to a Redis server.
+/// {@endtemplate}
+class RedisSocketOptions {
+  /// {@macro redis_socket_options}
+  const RedisSocketOptions({
+    this.host = 'localhost',
+    this.port = 6379,
+    this.timeout = const Duration(seconds: 30),
+  });
+
+  /// The host of the Redis server.
+  /// Defaults to localhost.
+  final String host;
+
+  /// The port of the Redis server.
+  /// Defaults to 6379.
+  final int port;
+
+  /// The timeout for connecting to the Redis server.
+  /// Defaults to 30 seconds.
+  final Duration timeout;
+}
+
+/// {@template redis_command_options}
+/// Options for sending commands to a Redis server.
+/// {@endtemplate}
+class RedisCommandOptions {
+  /// {@macro redis_command_options}
+  const RedisCommandOptions({
+    this.timeout = const Duration(seconds: 10),
+  });
+
+  /// The timeout for sending commands to the Redis server.
+  /// Defaults to 10 seconds.
+  final Duration timeout;
+}
+
 /// {@template redis_client}
 /// A client for interacting with a Redis server.
 /// {@endtemplate}
 class RedisClient {
   /// {@macro redis_client}
-  const RedisClient();
+  RedisClient({
+    RedisSocketOptions socket = const RedisSocketOptions(),
+    RedisCommandOptions command = const RedisCommandOptions(),
+  })  : _socketOptions = socket,
+        _commandOptions = command;
+
+  /// The socket options for the Redis server.
+  final RedisSocketOptions _socketOptions;
+
+  /// The command options for the Redis client.
+  final RedisCommandOptions _commandOptions;
+
+  /// The underlying connection to the Redis server.
+  RespServerConnection? _connection;
+
+  /// 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>();
+
+  /// 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 connected => _connected.future;
+
+  /// A future which completes when the client disconnects.
+  Future<void> get disconnected => _disconnected.future;
+
+  /// Authenticate to the Redis server.
+  /// Returns true if successful, otherwise false.
+  /// Equivalent to the `AUTH` command.
+  /// https://redis.io/commands/auth
+  Future<bool> 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;
+  }
+
+  /// Set the value of a key.
+  /// Equivalent to the `SET` command.
+  /// https://redis.io/commands/set
+  Future<void> set({required String key, required String value}) {
+    return _exec(() => RespCommandsTier2(_client!).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<String?> get({required String key}) {
+    return _exec(() => RespCommandsTier2(_client!).get(key));
+  }
+
+  /// Deletes the specified key.
+  /// Equivalent to the `DEL` command.
+  /// https://redis.io/commands/del
+  Future<void> delete({required String key}) {
+    return _exec(() => RespCommandsTier2(_client!).del([key]));
+  }
+
+  /// Send a command to the Redis server.
+  Future<RespType<dynamic>> sendCommand(List<Object?> command) async {
+    return _exec(() => RespCommandsTier0(_client!).execute(command));
+  }
+
+  /// Establish a connection to the Redis server.
+  /// The delay between connection attempts.
+  Future<void> connect({
+    Duration connectionRetryDelay = const Duration(milliseconds: 100),
+    int maxConnectionAttempts = 100,
+  }) async {
+    if (_closed) throw StateError('RedisClient has been closed.');
+
+    unawaited(
+      _reconnect(
+        connectionRetryDelay: connectionRetryDelay,
+        remainingConnectionAttempts: maxConnectionAttempts,
+      ),
+    );
+
+    return connected;
+  }
+
+  /// Terminate the connection to the Redis server.
+  Future<void> disconnect() {
+    _connection?.close();
+    _reset();
+    return disconnected;
+  }
+
+  /// 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() {
+    _closed = true;
+    return disconnect();
+  }
+
+  Future<void> _reconnect({
+    required Duration connectionRetryDelay,
+    required int remainingConnectionAttempts,
+    Object? error,
+    StackTrace? stackTrace,
+  }) async {
+    if (remainingConnectionAttempts <= 0) {
+      _connected.completeError(
+        error ?? const SocketException('Connection retry limit exceeded'),
+        stackTrace,
+      );
+      return;
+    }
+
+    void onConnectionOpened(RespServerConnection connection) {
+      _disconnected = Completer<void>();
+      _connection = connection;
+      _client = RespClient(connection);
+      _connected.complete();
+    }
+
+    void onConnectionClosed([Object? error, StackTrace? stackTrace]) {
+      _reset();
+      _reconnect(
+        connectionRetryDelay: connectionRetryDelay,
+        remainingConnectionAttempts: remainingConnectionAttempts - 1,
+        error: error,
+        stackTrace: stackTrace,
+      );
+    }
+
+    try {
+      final uri = _socketOptions.connectionUri;
+      final connection = await connectSocket(
+        uri.host,
+        port: uri.port,
+        timeout: _socketOptions.timeout,
+      );
+
+      onConnectionOpened(connection);
+
+      unawaited(
+        connection.outputSink.done
+            .then((_) => onConnectionClosed())
+            .catchError(onConnectionClosed),
+      );
+    } catch (error, stackTrace) {
+      Future<void>.delayed(
+        connectionRetryDelay,
+        () => _reconnect(
+          connectionRetryDelay: connectionRetryDelay,
+          remainingConnectionAttempts: remainingConnectionAttempts - 1,
+          error: error,
+          stackTrace: stackTrace,
+        ),
+      );
+    }
+  }
+
+  void _reset() {
+    _connected = Completer<void>();
+    _connection = null;
+    _client = null;
+    if (!_disconnected.isCompleted) _disconnected.complete();
+  }
+
+  Future<T> _exec<T>(FutureOr<T> Function() fn) async {
+    if (_closed) throw StateError('RedisClient has been closed.');
+    await connected;
+    return Future<T>.sync(() {
+      return fn();
+    }).timeout(
+      _commandOptions.timeout,
+      onTimeout: () {
+        _connection?.close();
+        throw const SocketException('Connection timed out');
+      },
+    );
+  }
+}
+
+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

@@ -6,6 +6,9 @@ publish_to: none
 environment:
   sdk: ">=3.0.0 <4.0.0"
 
+dependencies:
+  resp_client: ^1.2.0
+
 dev_dependencies:
   test: ^1.19.2
-  very_good_analysis: ^5.0.0
+  very_good_analysis: ^5.1.0

packages/redis_client/test/src/redis_client_test.dart

@@ -1,10 +1,159 @@
+import 'dart:io';
+
 import 'package:redis_client/redis_client.dart';
 import 'package:test/test.dart';
 
 void main() {
   group(RedisClient, () {
-    test('can be instantiated', () {
-      expect(const RedisClient(), isNotNull);
+    late RedisClient client;
+
+    setUp(() async {
+      client = RedisClient();
+      await client.connect();
+    });
+
+    tearDown(() async {
+      try {
+        await client.sendCommand(['RESET']);
+        await client.sendCommand(['FLUSHALL']);
+        await client.close();
+      } catch (_) {
+        // ignore
+      }
+    });
+
+    group('connect', () {
+      test('throws SocketException when connection times out', () async {
+        final client = RedisClient(
+          socket: const RedisSocketOptions(timeout: Duration(microseconds: 1)),
+        );
+        await expectLater(
+          () => client.connect(maxConnectionAttempts: 1),
+          throwsA(
+            isA<SocketException>().having(
+              (e) => e.message,
+              'message',
+              contains('Connection timed out'),
+            ),
+          ),
+        );
+      });
+
+      test('throws SocketException after max connection attempts', () async {
+        final client = RedisClient(
+          socket: const RedisSocketOptions(port: 1234),
+        );
+        await expectLater(
+          () => client.connect(maxConnectionAttempts: 1),
+          throwsA(
+            isA<SocketException>().having(
+              (e) => e.message,
+              'message',
+              contains('Connection refused'),
+            ),
+          ),
+        );
+      });
+
+      test('throws SocketException after disconnect', () async {
+        final client = RedisClient(
+          socket: const RedisSocketOptions(port: 1234),
+        );
+        await expectLater(
+          () => client.connect(maxConnectionAttempts: 0),
+          throwsA(
+            isA<SocketException>().having(
+              (e) => '$e',
+              'message',
+              contains('Connection retry limit exceeded'),
+            ),
+          ),
+        );
+      });
+
+      test('throws StateError when closed', () async {
+        await client.close();
+        await expectLater(
+          client.connect(),
+          throwsA(
+            isA<StateError>().having(
+              (e) => e.message,
+              'message',
+              'RedisClient has been closed.',
+            ),
+          ),
+        );
+      });
+    });
+
+    group('AUTH', () {
+      test('is required', () async {
+        await expectLater(
+          client.get(key: 'foo'),
+          throwsA(
+            isA<StateError>().having(
+              (e) => e.message,
+              'message',
+              contains('-NOAUTH Authentication required.'),
+            ),
+          ),
+        );
+      });
+
+      test('fails when username is incorrect', () async {
+        await expectLater(
+          client.auth(username: 'shorebird', password: 'password'),
+          completion(isFalse),
+        );
+      });
+
+      test('fails when password is incorrect', () async {
+        await expectLater(
+          client.auth(password: 'oops'),
+          completion(isFalse),
+        );
+      });
+
+      test('succeeds when username/password are correct', () async {
+        await expectLater(
+          client.auth(password: 'password'),
+          completion(isTrue),
+        );
+      });
+    });
+
+    group('GET/SET/DEL', () {
+      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)));
+        await expectLater(client.delete(key: key), completes);
+        await expectLater(client.get(key: key), completion(isNull));
+      });
+
+      test(
+          'throws SocketException '
+          '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<SocketException>().having(
+              (e) => e.message,
+              'message',
+              contains('Connection timed out'),
+            ),
+          ),
+        );
+      });
     });
   });
 }