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

feat(redis_client): add TS.ADD and TS.GET (#3036)

Browse Source
This commit is contained in:
Felix Angelov
2025-04-02 15:13:17 -05:00
committed by GitHub
parent 38728d258a
commit de9ba62cd3
3 changed files with 243 additions and 14 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/redis_client/lib/shorebird_redis_client.dart
+5 -1
View File
@@ -7,5 +7,9 @@ export 'src/redis_client.dart'
RedisLogger,
RedisSocketOptions,
RedisTimeSeries,
RedisTimeSeriesClientClock,
RedisTimeSeriesClock,
RedisTimeSeriesDuplicatePolicy,
RedisTimeSeriesEncoding;
RedisTimeSeriesEncoding,
RedisTimeSeriesServerClock,
RedisTimeSeriesTimestamp;
packages/redis_client/lib/src/redis_client.dart
+141 -1
View File
@@ -515,6 +515,87 @@ enum RedisTimeSeriesDuplicatePolicy {
String toArgument() => name.toUpperCase();
}
/// {@template redis_time_series_clock}
/// An abstract class which represents the clock
/// in the context of a redis time series instance.
/// {@endtemplate}
// ignore: one_member_abstracts
abstract class RedisTimeSeriesClock {
/// {@macro redis_time_series_clock}
const RedisTimeSeriesClock();
/// Returns a timestamp representing the current moment.
RedisTimeSeriesTimestamp now();
}
/// {@template redis_time_series_server_clock}
/// A [RedisTimeSeriesClock] that represents time on the server.
/// {@endtemplate}
class RedisTimeSeriesServerClock extends RedisTimeSeriesClock {
/// {@macro redis_time_series_server_clock}
const RedisTimeSeriesServerClock();
@override
RedisTimeSeriesTimestamp now() => const RedisTimeSeriesTimestamp._('*');
}
/// {@template redis_time_series_client_clock}
/// A [RedisTimeSeriesClock] that represents time on the client.
/// {@endtemplate}
class RedisTimeSeriesClientClock extends RedisTimeSeriesClock {
/// {@macro redis_time_series_client_clock}
const RedisTimeSeriesClientClock();
@override
RedisTimeSeriesTimestamp now() {
return RedisTimeSeriesTimestamp(DateTime.timestamp());
}
}
/// {@template redis_time_series_timestamp}
/// is Unix time (integer, in milliseconds) specifying the sample timestamp or *
/// to set the sample timestamp to the Unix time of the server's clock.
///
/// Unix time is the number of milliseconds that have elapsed since 00:00:00 UTC
/// on 1 January 1970, the Unix epoch, without adjustments made due to leap
/// seconds.
/// {@endtemplate}
class RedisTimeSeriesTimestamp {
/// Create a timestamp from a specific [dateTime].
///
/// See also:
/// * [client] for creating timestamps using the client clock.
/// * [server] for creating timestamps using the server clock.
/// {@macro redis_time_series_timestamp}
RedisTimeSeriesTimestamp(DateTime dateTime)
: this._('${dateTime.millisecondsSinceEpoch}');
const RedisTimeSeriesTimestamp._(this.value);
/// The client clock.
/// Useful for creating a timestamp using the client clock.
/// ```dart
/// await redis.timeSeries.add(
/// key: 'sensor',
/// timestamp: RedisTimeSeriesTimestamp.client.now(),
/// value: 42,
/// );
/// ```
static const RedisTimeSeriesClock client = RedisTimeSeriesClientClock();
/// The server clock.
/// Useful for creating a timestamp using the server clock.
/// ```dart
/// await redis.timeSeries.add(
/// key: 'sensor',
/// timestamp: RedisTimeSeriesTimestamp.server.now(),
/// value: 42,
/// );
/// ```
static const RedisTimeSeriesClock server = RedisTimeSeriesServerClock();
/// The underlying value of the timestamp.
final String value;
}
/// {@template redis_time_series}
/// An object that adds support for storing and querying timestamped data
/// points.
@@ -528,6 +609,7 @@ class RedisTimeSeries {
/// Create a new time series.
/// Equivalent to the `TS.CREATE` command.
/// https://redis.io/commands/ts.create
Future<void> create({
required String key,
Duration? retention,
@@ -535,7 +617,7 @@ class RedisTimeSeries {
int? chunkSize,
RedisTimeSeriesDuplicatePolicy? duplicatePolicy,
List<({String label, String value})>? labels,
}) async {
}) {
return _client.execute([
'TS.CREATE',
key,
@@ -552,6 +634,64 @@ class RedisTimeSeries {
],
]);
}
/// Append a sample to a time series.
/// Equivalent to the `TS.ADD` command.
/// Note: When the specified key does not exist, a new time series is created.
/// https://redis.io/commands/ts.add
Future<void> add({
required String key,
required RedisTimeSeriesTimestamp timestamp,
required double value,
Duration? retention,
RedisTimeSeriesEncoding? encoding,
int? chunkSize,
RedisTimeSeriesDuplicatePolicy? duplicatePolicy,
RedisTimeSeriesDuplicatePolicy? onDuplicate,
List<({String label, String value})>? labels,
}) {
return _client.execute([
'TS.ADD',
key,
timestamp.value,
value,
if (retention != null) ...['RETENTION', retention.inMilliseconds],
if (encoding != null) ...['ENCODING', encoding.toArgument()],
if (chunkSize != null) ...['CHUNK_SIZE', chunkSize],
if (duplicatePolicy != null) ...[
'DUPLICATE_POLICY',
duplicatePolicy.toArgument(),
],
if (onDuplicate != null) ...['ON_DUPLICATE', onDuplicate.toArgument()],
if (labels != null) ...[
'LABELS',
for (final label in labels) ...[label.label, label.value],
],
]);
}
/// Get the sample with the highest timestamp from a given time series.
/// Equivalent to the `TS.GET` command.
/// Returns a timestamp, value pair of the sample with the highest timestamp.
/// Throws a [RedisException] if the key does not exist.
/// Returns null if the time series is empty.
/// The returned timestamp will always be UTC.
/// https://redis.io/commands/ts.get
Future<({DateTime timestamp, double value})?> get({
required String key,
}) async {
final result = await _client.execute(['TS.GET', key]) as List<RespType>;
if (result.isEmpty) return null;
final timestamp = result[0] as RespInteger;
final value = result[1] as RespSimpleString;
return (
timestamp: DateTime.fromMillisecondsSinceEpoch(
timestamp.payload,
isUtc: true,
),
value: double.parse(value.payload),
);
}
}
extension on RedisSocketOptions {
packages/redis_client/test/src/redis_client_test.dart
+97 -12
View File
@@ -177,8 +177,7 @@ void main() {
tearDown(() async {
try {
await client.execute(['RESET']);
await client.execute(['FLUSHALL']);
await client.execute(['FLUSHALL SYNC']);
} on Exception {
// ignore
}
@@ -228,8 +227,7 @@ void main() {
tearDown(() async {
try {
await client.execute(['RESET']);
await client.execute(['FLUSHALL']);
await client.execute(['FLUSHALL SYNC']);
} on Exception {
// ignore
}
@@ -266,8 +264,7 @@ void main() {
for (final pair in kvPairs) {
await expectLater(client.delete(key: pair.key), completes);
}
await client.execute(['RESET']);
await client.execute(['FLUSHALL']);
await client.execute(['FLUSHALL SYNC']);
} on Exception {
// ignore
}
@@ -294,8 +291,7 @@ void main() {
tearDown(() async {
try {
await client.execute(['RESET']);
await client.execute(['FLUSHALL']);
await client.execute(['FLUSHALL SYNC']);
} on Exception {
// ignore
}
@@ -402,24 +398,27 @@ void main() {
});
group('TimeSeries', () {
group('CREATE', () {
group('CREATE/ADD/GET', () {
const key = 'sensor';
setUp(() async {
await client.connect();
await expectLater(client.delete(key: key), completes);
});
tearDown(() async {
try {
await client.execute(['RESET']);
await client.execute(['FLUSHALL']);
await client.execute(['FLUSHALL SYNC']);
} on Exception {
// ignore
}
});
test('completes', () async {
final date = DateTime(2025).toUtc();
await expectLater(
client.timeSeries.create(
key: 'sensor',
key: key,
chunkSize: 128,
duplicatePolicy: RedisTimeSeriesDuplicatePolicy.sum,
encoding: RedisTimeSeriesEncoding.compressed,
@@ -428,6 +427,92 @@ void main() {
),
completes,
);
await expectLater(
client.timeSeries.get(key: key),
completion(isNull), // Empty series
);
await expectLater(
client.timeSeries.add(
key: key,
timestamp: RedisTimeSeriesTimestamp(date),
value: 42,
chunkSize: 128,
duplicatePolicy: RedisTimeSeriesDuplicatePolicy.sum,
onDuplicate: RedisTimeSeriesDuplicatePolicy.sum,
encoding: RedisTimeSeriesEncoding.compressed,
retention: const Duration(days: 30),
labels: [(label: 'city', value: 'chicago')],
),
completes,
);
await expectLater(
client.timeSeries.get(key: key),
completion(equals((timestamp: date, value: 42))),
);
await expectLater(
client.timeSeries.add(
key: key,
timestamp: RedisTimeSeriesTimestamp(date),
value: 42,
chunkSize: 128,
duplicatePolicy: RedisTimeSeriesDuplicatePolicy.sum,
encoding: RedisTimeSeriesEncoding.compressed,
retention: const Duration(days: 30),
labels: [(label: 'city', value: 'chicago')],
),
completes,
);
await expectLater(
client.timeSeries.get(key: key),
completion(equals((timestamp: date, value: 84))),
);
await expectLater(
client.timeSeries.add(
key: key,
timestamp:
RedisTimeSeriesTimestamp.client.now(), // Use client clock
value: 56,
),
completes,
);
await expectLater(
client.timeSeries.get(key: key),
completion(
isA<({DateTime timestamp, double value})>()
.having(
(r) => r.timestamp.millisecondsSinceEpoch,
'timestamp',
closeTo(DateTime.timestamp().millisecondsSinceEpoch, 1000),
)
.having((r) => r.value, 'value', equals(56)),
),
);
await expectLater(
client.timeSeries.add(
key: key,
timestamp:
RedisTimeSeriesTimestamp.server.now(), // Use server clock
value: 99,
),
completes,
);
await expectLater(
client.timeSeries.get(key: key),
completion(
isA<({DateTime timestamp, double value})>()
.having(
(r) => r.timestamp.millisecondsSinceEpoch,
'timestamp',
closeTo(DateTime.timestamp().millisecondsSinceEpoch, 1000),
)
.having((r) => r.value, 'value', equals(99)),
),
);
await expectLater(client.delete(key: key), completes);
await expectLater(
client.timeSeries.get(key: key),
throwsA(isA<RedisException>()), // No key exists
);
});
});
});