mirror of
https://github.com/redis/go-redis.git
synced 2025-07-29 17:41:15 +03:00
Making command structs digestable (#2716)
* intial move * adding stringcmdable * moving module commands to align with other changes --------- Co-authored-by: ofekshenawa <104765379+ofekshenawa@users.noreply.github.com> Co-authored-by: ofekshenawa <ofek.shenawa@redis.com>
This commit is contained in:
Chayim
GitHub
920
timeseries_commands.go
Normal file
920
timeseries_commands.go
Normal file
@ -0,0 +1,920 @@
|
||||
package redis
|
||||
|
||||
import (
|
||||
"context"
|
||||
"strconv"
|
||||
|
||||
"github.com/redis/go-redis/v9/internal/proto"
|
||||
)
|
||||
|
||||
type TimeseriesCmdable interface {
|
||||
TSAdd(ctx context.Context, key string, timestamp interface{}, value float64) *IntCmd
|
||||
TSAddWithArgs(ctx context.Context, key string, timestamp interface{}, value float64, options *TSOptions) *IntCmd
|
||||
TSCreate(ctx context.Context, key string) *StatusCmd
|
||||
TSCreateWithArgs(ctx context.Context, key string, options *TSOptions) *StatusCmd
|
||||
TSAlter(ctx context.Context, key string, options *TSAlterOptions) *StatusCmd
|
||||
TSCreateRule(ctx context.Context, sourceKey string, destKey string, aggregator Aggregator, bucketDuration int) *StatusCmd
|
||||
TSCreateRuleWithArgs(ctx context.Context, sourceKey string, destKey string, aggregator Aggregator, bucketDuration int, options *TSCreateRuleOptions) *StatusCmd
|
||||
TSIncrBy(ctx context.Context, Key string, timestamp float64) *IntCmd
|
||||
TSIncrByWithArgs(ctx context.Context, key string, timestamp float64, options *TSIncrDecrOptions) *IntCmd
|
||||
TSDecrBy(ctx context.Context, Key string, timestamp float64) *IntCmd
|
||||
TSDecrByWithArgs(ctx context.Context, key string, timestamp float64, options *TSIncrDecrOptions) *IntCmd
|
||||
TSDel(ctx context.Context, Key string, fromTimestamp int, toTimestamp int) *IntCmd
|
||||
TSDeleteRule(ctx context.Context, sourceKey string, destKey string) *StatusCmd
|
||||
TSGet(ctx context.Context, key string) *TSTimestampValueCmd
|
||||
TSGetWithArgs(ctx context.Context, key string, options *TSGetOptions) *TSTimestampValueCmd
|
||||
TSInfo(ctx context.Context, key string) *MapStringInterfaceCmd
|
||||
TSInfoWithArgs(ctx context.Context, key string, options *TSInfoOptions) *MapStringInterfaceCmd
|
||||
TSMAdd(ctx context.Context, ktvSlices [][]interface{}) *IntSliceCmd
|
||||
TSQueryIndex(ctx context.Context, filterExpr []string) *StringSliceCmd
|
||||
TSRevRange(ctx context.Context, key string, fromTimestamp int, toTimestamp int) *TSTimestampValueSliceCmd
|
||||
TSRevRangeWithArgs(ctx context.Context, key string, fromTimestamp int, toTimestamp int, options *TSRevRangeOptions) *TSTimestampValueSliceCmd
|
||||
TSRange(ctx context.Context, key string, fromTimestamp int, toTimestamp int) *TSTimestampValueSliceCmd
|
||||
TSRangeWithArgs(ctx context.Context, key string, fromTimestamp int, toTimestamp int, options *TSRangeOptions) *TSTimestampValueSliceCmd
|
||||
TSMRange(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string) *MapStringSliceInterfaceCmd
|
||||
TSMRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRangeOptions) *MapStringSliceInterfaceCmd
|
||||
TSMRevRange(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string) *MapStringSliceInterfaceCmd
|
||||
TSMRevRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRevRangeOptions) *MapStringSliceInterfaceCmd
|
||||
TSMGet(ctx context.Context, filters []string) *MapStringSliceInterfaceCmd
|
||||
TSMGetWithArgs(ctx context.Context, filters []string, options *TSMGetOptions) *MapStringSliceInterfaceCmd
|
||||
}
|
||||
|
||||
type TSOptions struct {
|
||||
Retention int
|
||||
ChunkSize int
|
||||
Encoding string
|
||||
DuplicatePolicy string
|
||||
Labels map[string]string
|
||||
}
|
||||
type TSIncrDecrOptions struct {
|
||||
Timestamp int64
|
||||
Retention int
|
||||
ChunkSize int
|
||||
Uncompressed bool
|
||||
Labels map[string]string
|
||||
}
|
||||
|
||||
type TSAlterOptions struct {
|
||||
Retention int
|
||||
ChunkSize int
|
||||
DuplicatePolicy string
|
||||
Labels map[string]string
|
||||
}
|
||||
|
||||
type TSCreateRuleOptions struct {
|
||||
alignTimestamp int64
|
||||
}
|
||||
|
||||
type TSGetOptions struct {
|
||||
Latest bool
|
||||
}
|
||||
|
||||
type TSInfoOptions struct {
|
||||
Debug bool
|
||||
}
|
||||
type Aggregator int
|
||||
|
||||
const (
|
||||
Invalid = Aggregator(iota)
|
||||
Avg
|
||||
Sum
|
||||
Min
|
||||
Max
|
||||
Range
|
||||
Count
|
||||
First
|
||||
Last
|
||||
StdP
|
||||
StdS
|
||||
VarP
|
||||
VarS
|
||||
Twa
|
||||
)
|
||||
|
||||
func (a Aggregator) String() string {
|
||||
switch a {
|
||||
case Invalid:
|
||||
return ""
|
||||
case Avg:
|
||||
return "AVG"
|
||||
case Sum:
|
||||
return "SUM"
|
||||
case Min:
|
||||
return "MIN"
|
||||
case Max:
|
||||
return "MAX"
|
||||
case Range:
|
||||
return "RANGE"
|
||||
case Count:
|
||||
return "COUNT"
|
||||
case First:
|
||||
return "FIRST"
|
||||
case Last:
|
||||
return "LAST"
|
||||
case StdP:
|
||||
return "STD.P"
|
||||
case StdS:
|
||||
return "STD.S"
|
||||
case VarP:
|
||||
return "VAR.P"
|
||||
case VarS:
|
||||
return "VAR.S"
|
||||
case Twa:
|
||||
return "TWA"
|
||||
default:
|
||||
return ""
|
||||
}
|
||||
}
|
||||
|
||||
type TSRangeOptions struct {
|
||||
Latest bool
|
||||
FilterByTS []int
|
||||
FilterByValue []int
|
||||
Count int
|
||||
Align interface{}
|
||||
Aggregator Aggregator
|
||||
BucketDuration int
|
||||
BucketTimestamp interface{}
|
||||
Empty bool
|
||||
}
|
||||
|
||||
type TSRevRangeOptions struct {
|
||||
Latest bool
|
||||
FilterByTS []int
|
||||
FilterByValue []int
|
||||
Count int
|
||||
Align interface{}
|
||||
Aggregator Aggregator
|
||||
BucketDuration int
|
||||
BucketTimestamp interface{}
|
||||
Empty bool
|
||||
}
|
||||
|
||||
type TSMRangeOptions struct {
|
||||
Latest bool
|
||||
FilterByTS []int
|
||||
FilterByValue []int
|
||||
WithLabels bool
|
||||
SelectedLabels []interface{}
|
||||
Count int
|
||||
Align interface{}
|
||||
Aggregator Aggregator
|
||||
BucketDuration int
|
||||
BucketTimestamp interface{}
|
||||
Empty bool
|
||||
GroupByLabel interface{}
|
||||
Reducer interface{}
|
||||
}
|
||||
|
||||
type TSMRevRangeOptions struct {
|
||||
Latest bool
|
||||
FilterByTS []int
|
||||
FilterByValue []int
|
||||
WithLabels bool
|
||||
SelectedLabels []interface{}
|
||||
Count int
|
||||
Align interface{}
|
||||
Aggregator Aggregator
|
||||
BucketDuration int
|
||||
BucketTimestamp interface{}
|
||||
Empty bool
|
||||
GroupByLabel interface{}
|
||||
Reducer interface{}
|
||||
}
|
||||
|
||||
type TSMGetOptions struct {
|
||||
Latest bool
|
||||
WithLabels bool
|
||||
SelectedLabels []interface{}
|
||||
}
|
||||
|
||||
// TSAdd - Adds one or more observations to a t-digest sketch.
|
||||
// For more information - https://redis.io/commands/ts.add/
|
||||
func (c cmdable) TSAdd(ctx context.Context, key string, timestamp interface{}, value float64) *IntCmd {
|
||||
args := []interface{}{"TS.ADD", key, timestamp, value}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSAddWithArgs - Adds one or more observations to a t-digest sketch.
|
||||
// This function also allows for specifying additional options such as:
|
||||
// Retention, ChunkSize, Encoding, DuplicatePolicy and Labels.
|
||||
// For more information - https://redis.io/commands/ts.add/
|
||||
func (c cmdable) TSAddWithArgs(ctx context.Context, key string, timestamp interface{}, value float64, options *TSOptions) *IntCmd {
|
||||
args := []interface{}{"TS.ADD", key, timestamp, value}
|
||||
if options != nil {
|
||||
if options.Retention != 0 {
|
||||
args = append(args, "RETENTION", options.Retention)
|
||||
}
|
||||
if options.ChunkSize != 0 {
|
||||
args = append(args, "CHUNK_SIZE", options.ChunkSize)
|
||||
}
|
||||
if options.Encoding != "" {
|
||||
args = append(args, "ENCODING", options.Encoding)
|
||||
}
|
||||
|
||||
if options.DuplicatePolicy != "" {
|
||||
args = append(args, "DUPLICATE_POLICY", options.DuplicatePolicy)
|
||||
}
|
||||
if options.Labels != nil {
|
||||
args = append(args, "LABELS")
|
||||
for label, value := range options.Labels {
|
||||
args = append(args, label, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSCreate - Creates a new time-series key.
|
||||
// For more information - https://redis.io/commands/ts.create/
|
||||
func (c cmdable) TSCreate(ctx context.Context, key string) *StatusCmd {
|
||||
args := []interface{}{"TS.CREATE", key}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSCreateWithArgs - Creates a new time-series key with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Retention, ChunkSize, Encoding, DuplicatePolicy and Labels.
|
||||
// For more information - https://redis.io/commands/ts.create/
|
||||
func (c cmdable) TSCreateWithArgs(ctx context.Context, key string, options *TSOptions) *StatusCmd {
|
||||
args := []interface{}{"TS.CREATE", key}
|
||||
if options != nil {
|
||||
if options.Retention != 0 {
|
||||
args = append(args, "RETENTION", options.Retention)
|
||||
}
|
||||
if options.ChunkSize != 0 {
|
||||
args = append(args, "CHUNK_SIZE", options.ChunkSize)
|
||||
}
|
||||
if options.Encoding != "" {
|
||||
args = append(args, "ENCODING", options.Encoding)
|
||||
}
|
||||
|
||||
if options.DuplicatePolicy != "" {
|
||||
args = append(args, "DUPLICATE_POLICY", options.DuplicatePolicy)
|
||||
}
|
||||
if options.Labels != nil {
|
||||
args = append(args, "LABELS")
|
||||
for label, value := range options.Labels {
|
||||
args = append(args, label, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSAlter - Alters an existing time-series key with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Retention, ChunkSize and DuplicatePolicy.
|
||||
// For more information - https://redis.io/commands/ts.alter/
|
||||
func (c cmdable) TSAlter(ctx context.Context, key string, options *TSAlterOptions) *StatusCmd {
|
||||
args := []interface{}{"TS.ALTER", key}
|
||||
if options != nil {
|
||||
if options.Retention != 0 {
|
||||
args = append(args, "RETENTION", options.Retention)
|
||||
}
|
||||
if options.ChunkSize != 0 {
|
||||
args = append(args, "CHUNK_SIZE", options.ChunkSize)
|
||||
}
|
||||
if options.DuplicatePolicy != "" {
|
||||
args = append(args, "DUPLICATE_POLICY", options.DuplicatePolicy)
|
||||
}
|
||||
if options.Labels != nil {
|
||||
args = append(args, "LABELS")
|
||||
for label, value := range options.Labels {
|
||||
args = append(args, label, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSCreateRule - Creates a compaction rule from sourceKey to destKey.
|
||||
// For more information - https://redis.io/commands/ts.createrule/
|
||||
func (c cmdable) TSCreateRule(ctx context.Context, sourceKey string, destKey string, aggregator Aggregator, bucketDuration int) *StatusCmd {
|
||||
args := []interface{}{"TS.CREATERULE", sourceKey, destKey, "AGGREGATION", aggregator.String(), bucketDuration}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSCreateRuleWithArgs - Creates a compaction rule from sourceKey to destKey with additional option.
|
||||
// This function allows for specifying additional option such as:
|
||||
// alignTimestamp.
|
||||
// For more information - https://redis.io/commands/ts.createrule/
|
||||
func (c cmdable) TSCreateRuleWithArgs(ctx context.Context, sourceKey string, destKey string, aggregator Aggregator, bucketDuration int, options *TSCreateRuleOptions) *StatusCmd {
|
||||
args := []interface{}{"TS.CREATERULE", sourceKey, destKey, "AGGREGATION", aggregator.String(), bucketDuration}
|
||||
if options != nil {
|
||||
if options.alignTimestamp != 0 {
|
||||
args = append(args, options.alignTimestamp)
|
||||
}
|
||||
}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSIncrBy - Increments the value of a time-series key by the specified timestamp.
|
||||
// For more information - https://redis.io/commands/ts.incrby/
|
||||
func (c cmdable) TSIncrBy(ctx context.Context, Key string, timestamp float64) *IntCmd {
|
||||
args := []interface{}{"TS.INCRBY", Key, timestamp}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSIncrByWithArgs - Increments the value of a time-series key by the specified timestamp with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Timestamp, Retention, ChunkSize, Uncompressed and Labels.
|
||||
// For more information - https://redis.io/commands/ts.incrby/
|
||||
func (c cmdable) TSIncrByWithArgs(ctx context.Context, key string, timestamp float64, options *TSIncrDecrOptions) *IntCmd {
|
||||
args := []interface{}{"TS.INCRBY", key, timestamp}
|
||||
if options != nil {
|
||||
if options.Timestamp != 0 {
|
||||
args = append(args, "TIMESTAMP", options.Timestamp)
|
||||
}
|
||||
if options.Retention != 0 {
|
||||
args = append(args, "RETENTION", options.Retention)
|
||||
}
|
||||
if options.ChunkSize != 0 {
|
||||
args = append(args, "CHUNK_SIZE", options.ChunkSize)
|
||||
}
|
||||
if options.Uncompressed {
|
||||
args = append(args, "UNCOMPRESSED")
|
||||
}
|
||||
if options.Labels != nil {
|
||||
args = append(args, "LABELS")
|
||||
for label, value := range options.Labels {
|
||||
args = append(args, label, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSDecrBy - Decrements the value of a time-series key by the specified timestamp.
|
||||
// For more information - https://redis.io/commands/ts.decrby/
|
||||
func (c cmdable) TSDecrBy(ctx context.Context, Key string, timestamp float64) *IntCmd {
|
||||
args := []interface{}{"TS.DECRBY", Key, timestamp}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSDecrByWithArgs - Decrements the value of a time-series key by the specified timestamp with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Timestamp, Retention, ChunkSize, Uncompressed and Labels.
|
||||
// For more information - https://redis.io/commands/ts.decrby/
|
||||
func (c cmdable) TSDecrByWithArgs(ctx context.Context, key string, timestamp float64, options *TSIncrDecrOptions) *IntCmd {
|
||||
args := []interface{}{"TS.DECRBY", key, timestamp}
|
||||
if options != nil {
|
||||
if options.Timestamp != 0 {
|
||||
args = append(args, "TIMESTAMP", options.Timestamp)
|
||||
}
|
||||
if options.Retention != 0 {
|
||||
args = append(args, "RETENTION", options.Retention)
|
||||
}
|
||||
if options.ChunkSize != 0 {
|
||||
args = append(args, "CHUNK_SIZE", options.ChunkSize)
|
||||
}
|
||||
if options.Uncompressed {
|
||||
args = append(args, "UNCOMPRESSED")
|
||||
}
|
||||
if options.Labels != nil {
|
||||
args = append(args, "LABELS")
|
||||
for label, value := range options.Labels {
|
||||
args = append(args, label, value)
|
||||
}
|
||||
}
|
||||
}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSDel - Deletes a range of samples from a time-series key.
|
||||
// For more information - https://redis.io/commands/ts.del/
|
||||
func (c cmdable) TSDel(ctx context.Context, Key string, fromTimestamp int, toTimestamp int) *IntCmd {
|
||||
args := []interface{}{"TS.DEL", Key, fromTimestamp, toTimestamp}
|
||||
cmd := NewIntCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSDeleteRule - Deletes a compaction rule from sourceKey to destKey.
|
||||
// For more information - https://redis.io/commands/ts.deleterule/
|
||||
func (c cmdable) TSDeleteRule(ctx context.Context, sourceKey string, destKey string) *StatusCmd {
|
||||
args := []interface{}{"TS.DELETERULE", sourceKey, destKey}
|
||||
cmd := NewStatusCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSGetWithArgs - Gets the last sample of a time-series key with additional option.
|
||||
// This function allows for specifying additional option such as:
|
||||
// Latest.
|
||||
// For more information - https://redis.io/commands/ts.get/
|
||||
func (c cmdable) TSGetWithArgs(ctx context.Context, key string, options *TSGetOptions) *TSTimestampValueCmd {
|
||||
args := []interface{}{"TS.GET", key}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
}
|
||||
cmd := newTSTimestampValueCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSGet - Gets the last sample of a time-series key.
|
||||
// For more information - https://redis.io/commands/ts.get/
|
||||
func (c cmdable) TSGet(ctx context.Context, key string) *TSTimestampValueCmd {
|
||||
args := []interface{}{"TS.GET", key}
|
||||
cmd := newTSTimestampValueCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
type TSTimestampValue struct {
|
||||
Timestamp int64
|
||||
Value float64
|
||||
}
|
||||
type TSTimestampValueCmd struct {
|
||||
baseCmd
|
||||
val TSTimestampValue
|
||||
}
|
||||
|
||||
func newTSTimestampValueCmd(ctx context.Context, args ...interface{}) *TSTimestampValueCmd {
|
||||
return &TSTimestampValueCmd{
|
||||
baseCmd: baseCmd{
|
||||
ctx: ctx,
|
||||
args: args,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueCmd) String() string {
|
||||
return cmdString(cmd, cmd.val)
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueCmd) SetVal(val TSTimestampValue) {
|
||||
cmd.val = val
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueCmd) Result() (TSTimestampValue, error) {
|
||||
return cmd.val, cmd.err
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueCmd) Val() TSTimestampValue {
|
||||
return cmd.val
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueCmd) readReply(rd *proto.Reader) (err error) {
|
||||
n, err := rd.ReadMapLen()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cmd.val = TSTimestampValue{}
|
||||
for i := 0; i < n; i++ {
|
||||
timestamp, err := rd.ReadInt()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
value, err := rd.ReadString()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cmd.val.Timestamp = timestamp
|
||||
cmd.val.Value, err = strconv.ParseFloat(value, 64)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// TSInfo - Returns information about a time-series key.
|
||||
// For more information - https://redis.io/commands/ts.info/
|
||||
func (c cmdable) TSInfo(ctx context.Context, key string) *MapStringInterfaceCmd {
|
||||
args := []interface{}{"TS.INFO", key}
|
||||
cmd := NewMapStringInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSInfoWithArgs - Returns information about a time-series key with additional option.
|
||||
// This function allows for specifying additional option such as:
|
||||
// Debug.
|
||||
// For more information - https://redis.io/commands/ts.info/
|
||||
func (c cmdable) TSInfoWithArgs(ctx context.Context, key string, options *TSInfoOptions) *MapStringInterfaceCmd {
|
||||
args := []interface{}{"TS.INFO", key}
|
||||
if options != nil {
|
||||
if options.Debug {
|
||||
args = append(args, "DEBUG")
|
||||
}
|
||||
}
|
||||
cmd := NewMapStringInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMAdd - Adds multiple samples to multiple time-series keys.
|
||||
// For more information - https://redis.io/commands/ts.madd/
|
||||
func (c cmdable) TSMAdd(ctx context.Context, ktvSlices [][]interface{}) *IntSliceCmd {
|
||||
args := []interface{}{"TS.MADD"}
|
||||
for _, ktv := range ktvSlices {
|
||||
args = append(args, ktv...)
|
||||
}
|
||||
cmd := NewIntSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSQueryIndex - Returns all the keys matching the filter expression.
|
||||
// For more information - https://redis.io/commands/ts.queryindex/
|
||||
func (c cmdable) TSQueryIndex(ctx context.Context, filterExpr []string) *StringSliceCmd {
|
||||
args := []interface{}{"TS.QUERYINDEX"}
|
||||
for _, f := range filterExpr {
|
||||
args = append(args, f)
|
||||
}
|
||||
cmd := NewStringSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSRevRange - Returns a range of samples from a time-series key in reverse order.
|
||||
// For more information - https://redis.io/commands/ts.revrange/
|
||||
func (c cmdable) TSRevRange(ctx context.Context, key string, fromTimestamp int, toTimestamp int) *TSTimestampValueSliceCmd {
|
||||
args := []interface{}{"TS.REVRANGE", key, fromTimestamp, toTimestamp}
|
||||
cmd := newTSTimestampValueSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSRevRangeWithArgs - Returns a range of samples from a time-series key in reverse order with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Latest, FilterByTS, FilterByValue, Count, Align, Aggregator,
|
||||
// BucketDuration, BucketTimestamp and Empty.
|
||||
// For more information - https://redis.io/commands/ts.revrange/
|
||||
func (c cmdable) TSRevRangeWithArgs(ctx context.Context, key string, fromTimestamp int, toTimestamp int, options *TSRevRangeOptions) *TSTimestampValueSliceCmd {
|
||||
args := []interface{}{"TS.REVRANGE", key, fromTimestamp, toTimestamp}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
if options.FilterByTS != nil {
|
||||
args = append(args, "FILTER_BY_TS")
|
||||
for _, f := range options.FilterByTS {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.FilterByValue != nil {
|
||||
args = append(args, "FILTER_BY_VALUE")
|
||||
for _, f := range options.FilterByValue {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.Count != 0 {
|
||||
args = append(args, "COUNT", options.Count)
|
||||
}
|
||||
if options.Align != nil {
|
||||
args = append(args, "ALIGN", options.Align)
|
||||
}
|
||||
if options.Aggregator != 0 {
|
||||
args = append(args, "AGGREGATION", options.Aggregator.String())
|
||||
}
|
||||
if options.BucketDuration != 0 {
|
||||
args = append(args, options.BucketDuration)
|
||||
}
|
||||
if options.BucketTimestamp != nil {
|
||||
args = append(args, "BUCKETTIMESTAMP", options.BucketTimestamp)
|
||||
}
|
||||
if options.Empty {
|
||||
args = append(args, "EMPTY")
|
||||
}
|
||||
}
|
||||
cmd := newTSTimestampValueSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSRange - Returns a range of samples from a time-series key.
|
||||
// For more information - https://redis.io/commands/ts.range/
|
||||
func (c cmdable) TSRange(ctx context.Context, key string, fromTimestamp int, toTimestamp int) *TSTimestampValueSliceCmd {
|
||||
args := []interface{}{"TS.RANGE", key, fromTimestamp, toTimestamp}
|
||||
cmd := newTSTimestampValueSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSRangeWithArgs - Returns a range of samples from a time-series key with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Latest, FilterByTS, FilterByValue, Count, Align, Aggregator,
|
||||
// BucketDuration, BucketTimestamp and Empty.
|
||||
// For more information - https://redis.io/commands/ts.range/
|
||||
func (c cmdable) TSRangeWithArgs(ctx context.Context, key string, fromTimestamp int, toTimestamp int, options *TSRangeOptions) *TSTimestampValueSliceCmd {
|
||||
args := []interface{}{"TS.RANGE", key, fromTimestamp, toTimestamp}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
if options.FilterByTS != nil {
|
||||
args = append(args, "FILTER_BY_TS")
|
||||
for _, f := range options.FilterByTS {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.FilterByValue != nil {
|
||||
args = append(args, "FILTER_BY_VALUE")
|
||||
for _, f := range options.FilterByValue {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.Count != 0 {
|
||||
args = append(args, "COUNT", options.Count)
|
||||
}
|
||||
if options.Align != nil {
|
||||
args = append(args, "ALIGN", options.Align)
|
||||
}
|
||||
if options.Aggregator != 0 {
|
||||
args = append(args, "AGGREGATION", options.Aggregator.String())
|
||||
}
|
||||
if options.BucketDuration != 0 {
|
||||
args = append(args, options.BucketDuration)
|
||||
}
|
||||
if options.BucketTimestamp != nil {
|
||||
args = append(args, "BUCKETTIMESTAMP", options.BucketTimestamp)
|
||||
}
|
||||
if options.Empty {
|
||||
args = append(args, "EMPTY")
|
||||
}
|
||||
}
|
||||
cmd := newTSTimestampValueSliceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
type TSTimestampValueSliceCmd struct {
|
||||
baseCmd
|
||||
val []TSTimestampValue
|
||||
}
|
||||
|
||||
func newTSTimestampValueSliceCmd(ctx context.Context, args ...interface{}) *TSTimestampValueSliceCmd {
|
||||
return &TSTimestampValueSliceCmd{
|
||||
baseCmd: baseCmd{
|
||||
ctx: ctx,
|
||||
args: args,
|
||||
},
|
||||
}
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueSliceCmd) String() string {
|
||||
return cmdString(cmd, cmd.val)
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueSliceCmd) SetVal(val []TSTimestampValue) {
|
||||
cmd.val = val
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueSliceCmd) Result() ([]TSTimestampValue, error) {
|
||||
return cmd.val, cmd.err
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueSliceCmd) Val() []TSTimestampValue {
|
||||
return cmd.val
|
||||
}
|
||||
|
||||
func (cmd *TSTimestampValueSliceCmd) readReply(rd *proto.Reader) (err error) {
|
||||
n, err := rd.ReadArrayLen()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cmd.val = make([]TSTimestampValue, n)
|
||||
for i := 0; i < n; i++ {
|
||||
_, _ = rd.ReadArrayLen()
|
||||
timestamp, err := rd.ReadInt()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
value, err := rd.ReadString()
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
cmd.val[i].Timestamp = timestamp
|
||||
cmd.val[i].Value, err = strconv.ParseFloat(value, 64)
|
||||
if err != nil {
|
||||
return err
|
||||
}
|
||||
}
|
||||
|
||||
return nil
|
||||
}
|
||||
|
||||
// TSMRange - Returns a range of samples from multiple time-series keys.
|
||||
// For more information - https://redis.io/commands/ts.mrange/
|
||||
func (c cmdable) TSMRange(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MRANGE", fromTimestamp, toTimestamp, "FILTER"}
|
||||
for _, f := range filterExpr {
|
||||
args = append(args, f)
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMRangeWithArgs - Returns a range of samples from multiple time-series keys with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Latest, FilterByTS, FilterByValue, WithLabels, SelectedLabels,
|
||||
// Count, Align, Aggregator, BucketDuration, BucketTimestamp,
|
||||
// Empty, GroupByLabel and Reducer.
|
||||
// For more information - https://redis.io/commands/ts.mrange/
|
||||
func (c cmdable) TSMRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRangeOptions) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MRANGE", fromTimestamp, toTimestamp}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
if options.FilterByTS != nil {
|
||||
args = append(args, "FILTER_BY_TS")
|
||||
for _, f := range options.FilterByTS {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.FilterByValue != nil {
|
||||
args = append(args, "FILTER_BY_VALUE")
|
||||
for _, f := range options.FilterByValue {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.WithLabels {
|
||||
args = append(args, "WITHLABELS")
|
||||
}
|
||||
if options.SelectedLabels != nil {
|
||||
args = append(args, "SELECTED_LABELS")
|
||||
args = append(args, options.SelectedLabels...)
|
||||
}
|
||||
if options.Count != 0 {
|
||||
args = append(args, "COUNT", options.Count)
|
||||
}
|
||||
if options.Align != nil {
|
||||
args = append(args, "ALIGN", options.Align)
|
||||
}
|
||||
if options.Aggregator != 0 {
|
||||
args = append(args, "AGGREGATION", options.Aggregator.String())
|
||||
}
|
||||
if options.BucketDuration != 0 {
|
||||
args = append(args, options.BucketDuration)
|
||||
}
|
||||
if options.BucketTimestamp != nil {
|
||||
args = append(args, "BUCKETTIMESTAMP", options.BucketTimestamp)
|
||||
}
|
||||
if options.Empty {
|
||||
args = append(args, "EMPTY")
|
||||
}
|
||||
}
|
||||
args = append(args, "FILTER")
|
||||
for _, f := range filterExpr {
|
||||
args = append(args, f)
|
||||
}
|
||||
if options != nil {
|
||||
if options.GroupByLabel != nil {
|
||||
args = append(args, "GROUPBY", options.GroupByLabel)
|
||||
}
|
||||
if options.Reducer != nil {
|
||||
args = append(args, "REDUCE", options.Reducer)
|
||||
}
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMRevRange - Returns a range of samples from multiple time-series keys in reverse order.
|
||||
// For more information - https://redis.io/commands/ts.mrevrange/
|
||||
func (c cmdable) TSMRevRange(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MREVRANGE", fromTimestamp, toTimestamp, "FILTER"}
|
||||
for _, f := range filterExpr {
|
||||
args = append(args, f)
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMRevRangeWithArgs - Returns a range of samples from multiple time-series keys in reverse order with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Latest, FilterByTS, FilterByValue, WithLabels, SelectedLabels,
|
||||
// Count, Align, Aggregator, BucketDuration, BucketTimestamp,
|
||||
// Empty, GroupByLabel and Reducer.
|
||||
// For more information - https://redis.io/commands/ts.mrevrange/
|
||||
func (c cmdable) TSMRevRangeWithArgs(ctx context.Context, fromTimestamp int, toTimestamp int, filterExpr []string, options *TSMRevRangeOptions) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MREVRANGE", fromTimestamp, toTimestamp}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
if options.FilterByTS != nil {
|
||||
args = append(args, "FILTER_BY_TS")
|
||||
for _, f := range options.FilterByTS {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.FilterByValue != nil {
|
||||
args = append(args, "FILTER_BY_VALUE")
|
||||
for _, f := range options.FilterByValue {
|
||||
args = append(args, f)
|
||||
}
|
||||
}
|
||||
if options.WithLabels {
|
||||
args = append(args, "WITHLABELS")
|
||||
}
|
||||
if options.SelectedLabels != nil {
|
||||
args = append(args, "SELECTED_LABELS")
|
||||
args = append(args, options.SelectedLabels...)
|
||||
}
|
||||
if options.Count != 0 {
|
||||
args = append(args, "COUNT", options.Count)
|
||||
}
|
||||
if options.Align != nil {
|
||||
args = append(args, "ALIGN", options.Align)
|
||||
}
|
||||
if options.Aggregator != 0 {
|
||||
args = append(args, "AGGREGATION", options.Aggregator.String())
|
||||
}
|
||||
if options.BucketDuration != 0 {
|
||||
args = append(args, options.BucketDuration)
|
||||
}
|
||||
if options.BucketTimestamp != nil {
|
||||
args = append(args, "BUCKETTIMESTAMP", options.BucketTimestamp)
|
||||
}
|
||||
if options.Empty {
|
||||
args = append(args, "EMPTY")
|
||||
}
|
||||
}
|
||||
args = append(args, "FILTER")
|
||||
for _, f := range filterExpr {
|
||||
args = append(args, f)
|
||||
}
|
||||
if options != nil {
|
||||
if options.GroupByLabel != nil {
|
||||
args = append(args, "GROUPBY", options.GroupByLabel)
|
||||
}
|
||||
if options.Reducer != nil {
|
||||
args = append(args, "REDUCE", options.Reducer)
|
||||
}
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMGet - Returns the last sample of multiple time-series keys.
|
||||
// For more information - https://redis.io/commands/ts.mget/
|
||||
func (c cmdable) TSMGet(ctx context.Context, filters []string) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MGET", "FILTER"}
|
||||
for _, f := range filters {
|
||||
args = append(args, f)
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
|
||||
// TSMGetWithArgs - Returns the last sample of multiple time-series keys with additional options.
|
||||
// This function allows for specifying additional options such as:
|
||||
// Latest, WithLabels and SelectedLabels.
|
||||
// For more information - https://redis.io/commands/ts.mget/
|
||||
func (c cmdable) TSMGetWithArgs(ctx context.Context, filters []string, options *TSMGetOptions) *MapStringSliceInterfaceCmd {
|
||||
args := []interface{}{"TS.MGET"}
|
||||
if options != nil {
|
||||
if options.Latest {
|
||||
args = append(args, "LATEST")
|
||||
}
|
||||
if options.WithLabels {
|
||||
args = append(args, "WITHLABELS")
|
||||
}
|
||||
if options.SelectedLabels != nil {
|
||||
args = append(args, "SELECTED_LABELS")
|
||||
args = append(args, options.SelectedLabels...)
|
||||
}
|
||||
}
|
||||
args = append(args, "FILTER")
|
||||
for _, f := range filters {
|
||||
args = append(args, f)
|
||||
}
|
||||
cmd := NewMapStringSliceInterfaceCmd(ctx, args...)
|
||||
_ = c(ctx, cmd)
|
||||
return cmd
|
||||
}
|
||||
Reference in New Issue
Block a user