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

feat(redis_client): add TS.RANGE (#3037)

Browse Source
This commit is contained in:
Felix Angelov
2025-04-02 16:38:56 -05:00
committed by GitHub
parent de9ba62cd3
commit bbd1754fce
3 changed files with 380 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
packages/redis_client/lib/shorebird_redis_client.dart
+6 -1
View File
@@ -7,9 +7,14 @@ export 'src/redis_client.dart'
RedisLogger,
RedisSocketOptions,
RedisTimeSeries,
RedisTimeSeriesAggregation,
RedisTimeSeriesAggregator,
RedisTimeSeriesAlign,
RedisTimeSeriesClientClock,
RedisTimeSeriesClock,
RedisTimeSeriesDuplicatePolicy,
RedisTimeSeriesEncoding,
RedisTimeSeriesFromTimestamp,
RedisTimeSeriesServerClock,
RedisTimeSeriesTimestamp;
RedisTimeSeriesTimestamp,
RedisTimeSeriesToTimestamp;
packages/redis_client/lib/src/redis_client.dart
+226 -35
View File
@@ -515,41 +515,6 @@ 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.
@@ -596,6 +561,181 @@ class RedisTimeSeriesTimestamp {
final String value;
}
/// {@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_from_timestamp}
/// The start timestamp for the range query (integer Unix timestamp in
/// milliseconds).
/// {@endtemplate}
class RedisTimeSeriesFromTimestamp {
/// {@macro redis_time_series_from_timestamp}
RedisTimeSeriesFromTimestamp(DateTime dateTime)
: this._('${dateTime.millisecondsSinceEpoch}');
/// The timestamp of the earliest sample among all the time series
/// that passes the provided filter.
const RedisTimeSeriesFromTimestamp.start() : value = '-';
const RedisTimeSeriesFromTimestamp._(this.value);
/// The underlying value of the timestamp.
final String value;
}
/// {@template redis_time_series_to_timestamp}
/// The end timestamp for the range query (integer Unix timestamp in
/// milliseconds).
/// {@endtemplate}
class RedisTimeSeriesToTimestamp {
/// {@macro redis_time_series_to_timestamp}
RedisTimeSeriesToTimestamp(DateTime dateTime)
: this._('${dateTime.millisecondsSinceEpoch}');
/// The timestamp of the latest sample among all the time series that passes
/// the provided filter.
const RedisTimeSeriesToTimestamp.end() : value = '+';
const RedisTimeSeriesToTimestamp._(this.value);
/// The underlying value of the timestamp.
final String value;
}
/// {@template redis_time_series_align}
/// The time bucket alignment control for AGGREGATION. It controls the time
/// bucket timestamps by changing the reference timestamp on which a bucket is
/// defined.
/// {@endtemplate}
class RedisTimeSeriesAlign {
/// A specific timestamp: align the reference timestamp to a specific time.
RedisTimeSeriesAlign(DateTime date)
: this._('${date.millisecondsSinceEpoch}');
/// The reference timestamp will be the query start interval time
/// (fromTimestamp) which can't be -.
RedisTimeSeriesAlign.start() : this._('-');
/// The reference timestamp will be the query end interval time (toTimestamp)
/// which can't be +.
RedisTimeSeriesAlign.end() : this._('+');
RedisTimeSeriesAlign._(this.value);
/// The underlying value of the alignment.
final String value;
}
enum RedisTimeSeriesAggregator {
/// Arithmetic mean of all values
average,
/// Sum of all values
sum,
/// Minimum value
min,
/// Maximum value
max,
/// Difference between the maximum and the minimum value
range,
/// Number of values
count,
/// Value with lowest timestamp in the bucket
first,
/// Value with highest timestamp in the bucket
last,
/// Population standard deviation of the values
populationStandardDeviation,
/// Sample standard deviation of the values
sampleStandardDeviation,
/// Population variance of the values
populationVariance,
/// Sample variance of the values
sampleVariance,
/// Time-weighted average over the bucket's timeframe
timeWeightedAverage;
/// Converts the enum to an argument that can be passed directly to
/// `execute`.
String toArgument() {
return switch (this) {
RedisTimeSeriesAggregator.average => 'avg',
RedisTimeSeriesAggregator.sum => 'sum',
RedisTimeSeriesAggregator.min => 'min',
RedisTimeSeriesAggregator.max => 'max',
RedisTimeSeriesAggregator.range => 'range',
RedisTimeSeriesAggregator.count => 'count',
RedisTimeSeriesAggregator.first => 'first',
RedisTimeSeriesAggregator.last => 'last',
RedisTimeSeriesAggregator.populationStandardDeviation => 'std.p',
RedisTimeSeriesAggregator.sampleStandardDeviation => 'std.s',
RedisTimeSeriesAggregator.populationVariance => 'var.p',
RedisTimeSeriesAggregator.sampleVariance => 'var.s',
RedisTimeSeriesAggregator.timeWeightedAverage => 'twa',
};
}
}
/// {@template redis_time_series_aggregation}
/// Aggregates time series samples into time buckets.
/// {@endtemplate}
class RedisTimeSeriesAggregation {
/// {@macro redis_time_series_aggregation}
const RedisTimeSeriesAggregation({
required this.aggregator,
required this.bucketDuration,
});
/// The aggregation type.
final RedisTimeSeriesAggregator aggregator;
/// The duration of each bucket.
final Duration bucketDuration;
}
/// {@template redis_time_series}
/// An object that adds support for storing and querying timestamped data
/// points.
@@ -692,6 +832,57 @@ class RedisTimeSeries {
value: double.parse(value.payload),
);
}
/// Query a range in forward direction.
/// Equivalent to the `TS.RANGE` command.
/// https://redis.io/commands/ts.range
Future<List<({DateTime timestamp, double value})>> range({
required String key,
required RedisTimeSeriesFromTimestamp from,
required RedisTimeSeriesToTimestamp to,
List<RedisTimeSeriesTimestamp>? filterByTimestamp,
({double min, double max})? filterByValue,
int? count,
RedisTimeSeriesAlign? align,
RedisTimeSeriesAggregation? aggregation,
}) async {
final results =
await _client.execute([
'TS.RANGE',
key,
from.value,
to.value,
if (filterByTimestamp != null) ...[
'FILTER_BY_TS',
...filterByTimestamp.map((t) => t.value),
],
if (filterByValue != null) ...[
'FILTER_BY_VALUE',
filterByValue.min,
filterByValue.max,
],
if (count != null) ...['COUNT', count],
if (align != null) ...['ALIGN', align.value],
if (aggregation != null) ...[
'AGGREGATION',
aggregation.aggregator.toArgument(),
aggregation.bucketDuration.inMilliseconds,
],
])
as List<RespType>;
return results.map((result) {
final payload = result.payload as List<RespType>;
final timestamp = payload[0] as RespInteger;
final value = payload[1] as RespSimpleString;
return (
timestamp: DateTime.fromMillisecondsSinceEpoch(
timestamp.payload,
isUtc: true,
),
value: double.parse(value.payload),
);
}).toList();
}
}
extension on RedisSocketOptions {
packages/redis_client/test/src/redis_client_test.dart
+148
View File
@@ -515,6 +515,154 @@ void main() {
);
});
});
group('RANGE', () {
const key = 'sensor';
final data = [
(timestamp: DateTime(2020).toUtc(), value: 1.0),
(timestamp: DateTime(2021).toUtc(), value: 2.0),
(timestamp: DateTime(2022).toUtc(), value: 3.0),
(timestamp: DateTime(2023).toUtc(), value: 4.0),
];
setUp(() async {
await client.connect();
await expectLater(client.delete(key: key), completes);
for (final tuple in data) {
await expectLater(
client.timeSeries.add(
key: key,
timestamp: RedisTimeSeriesTimestamp(tuple.timestamp),
value: tuple.value,
),
completes,
);
}
});
test('completes', () async {
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: const RedisTimeSeriesToTimestamp.end(),
),
completion(containsAllInOrder(data)),
);
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: const RedisTimeSeriesToTimestamp.end(),
count: 3,
),
completion(containsAllInOrder(data.sublist(0, 2))),
);
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: const RedisTimeSeriesToTimestamp.end(),
filterByTimestamp: [
RedisTimeSeriesTimestamp(DateTime(2023).toUtc()),
],
),
completion(containsAllInOrder([data.last])),
);
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: const RedisTimeSeriesToTimestamp.end(),
filterByValue: (min: 0, max: 1),
),
completion(containsAllInOrder(data.sublist(0, 1))),
);
await expectLater(
client.timeSeries.range(
key: key,
from: RedisTimeSeriesFromTimestamp(data.first.timestamp),
to: const RedisTimeSeriesToTimestamp.end(),
aggregation: const RedisTimeSeriesAggregation(
aggregator: RedisTimeSeriesAggregator.timeWeightedAverage,
bucketDuration: Duration(days: 365 * 3),
),
align: RedisTimeSeriesAlign.start(),
),
completion(
containsAllInOrder([
(timestamp: DateTime(2020).toUtc(), value: 2.4977181459936197),
(
timestamp: DateTime(2022, 12, 31).toUtc(),
value: 3.998630136986301,
),
]),
),
);
await expectLater(
client.timeSeries.range(
key: key,
from: RedisTimeSeriesFromTimestamp(data.first.timestamp),
to: const RedisTimeSeriesToTimestamp.end(),
aggregation: const RedisTimeSeriesAggregation(
aggregator: RedisTimeSeriesAggregator.count,
bucketDuration: Duration(days: 365 * 3),
),
align: RedisTimeSeriesAlign.start(),
),
completion(
containsAllInOrder([
(timestamp: DateTime(2020).toUtc(), value: 3.0),
(timestamp: DateTime(2022, 12, 31).toUtc(), value: 1.0),
]),
),
);
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: RedisTimeSeriesToTimestamp(data.last.timestamp),
aggregation: const RedisTimeSeriesAggregation(
aggregator: RedisTimeSeriesAggregator.count,
bucketDuration: Duration(days: 365 * 3),
),
align: RedisTimeSeriesAlign.end(),
),
completion(
containsAllInOrder([
(timestamp: DateTime(2017, 1, 2).toUtc(), value: 1.0),
(timestamp: DateTime(2020, 1, 2).toUtc(), value: 2.0),
(timestamp: DateTime(2023).toUtc(), value: 1.0),
]),
),
);
await expectLater(
client.timeSeries.range(
key: key,
from: const RedisTimeSeriesFromTimestamp.start(),
to: RedisTimeSeriesToTimestamp(data.last.timestamp),
aggregation: const RedisTimeSeriesAggregation(
aggregator: RedisTimeSeriesAggregator.count,
bucketDuration: Duration(days: 365 * 3),
),
align: RedisTimeSeriesAlign(DateTime(2022).toUtc()),
),
completion(
containsAllInOrder([
(timestamp: DateTime(2019, 1, 2).toUtc(), value: 2.0),
(timestamp: DateTime(2022).toUtc(), value: 2.0),
]),
),
);
});
});
});
});
}