-
Notifications
You must be signed in to change notification settings - Fork 14
Commit
This commit does not belong to any branch on this repository, and may belong to a fork outside of the repository.
feat(transform): Add UtilityControl Transform for Generating Ctrl Mes…
…sages (#175) * feat(transform): Add UtilityControl Transform for Generating Ctrl Messages * fix(transform): Fix Cases Where Aggregate Transforms Emit Empty Results * docs(examples): Add Example for Generating Ctrl Messages
- Loading branch information
Showing
6 changed files
with
125 additions
and
0 deletions.
There are no files selected for viewing
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
13 changes: 13 additions & 0 deletions
13
examples/config/transform/utility/generate_ctrl/config.jsonnet
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,13 @@ | ||
// This example shows how to use the `utility_control` transform to | ||
// generate a control (ctrl) Message based on the amount of data Messages | ||
// received by the system. ctrl Messages overrides the settings of the | ||
// `aggregate_to_array` transform (and any other transform that supports). | ||
local sub = import '../../../../../build/config/substation.libsonnet'; | ||
|
||
{ | ||
transforms: [ | ||
sub.tf.utility.control({ batch: { count: 2 } }), | ||
sub.tf.aggregate.to.array({ batch: { count: 10000 } }), | ||
sub.tf.send.stdout(), | ||
], | ||
} |
13 changes: 13 additions & 0 deletions
13
examples/config/transform/utility/generate_ctrl/data.jsonl
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,13 @@ | ||
{"a":"b"} | ||
{"c":"d"} | ||
{"e":"f"} | ||
{"g":"h"} | ||
{"i":"j"} | ||
{"k":"l"} | ||
{"m":"n"} | ||
{"o":"p"} | ||
{"q":"r"} | ||
{"s":"t"} | ||
{"u":"v"} | ||
{"w":"x"} | ||
{"y":"z"} |
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
This file contains bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Original file line number | Diff line number | Diff line change |
---|---|---|
@@ -0,0 +1,81 @@ | ||
package transform | ||
|
||
import ( | ||
"context" | ||
"encoding/json" | ||
"fmt" | ||
"sync" | ||
|
||
"github.com/brexhq/substation/config" | ||
"github.com/brexhq/substation/internal/aggregate" | ||
iconfig "github.com/brexhq/substation/internal/config" | ||
"github.com/brexhq/substation/message" | ||
) | ||
|
||
type utilityControlConfig struct { | ||
Batch iconfig.Batch `json:"batch"` | ||
} | ||
|
||
func (c *utilityControlConfig) Decode(in interface{}) error { | ||
return iconfig.Decode(in, c) | ||
} | ||
|
||
func newUtilityControl(_ context.Context, cfg config.Config) (*utilityControl, error) { | ||
conf := utilityControlConfig{} | ||
if err := conf.Decode(cfg.Settings); err != nil { | ||
return nil, fmt.Errorf("transform: utility_control: %v", err) | ||
} | ||
|
||
agg, err := aggregate.New(aggregate.Config{ | ||
Count: conf.Batch.Count, | ||
Size: conf.Batch.Size, | ||
Duration: conf.Batch.Duration, | ||
}) | ||
if err != nil { | ||
return nil, fmt.Errorf("transform: utility_control: %v", err) | ||
} | ||
|
||
tf := utilityControl{ | ||
conf: conf, | ||
agg: *agg, | ||
} | ||
|
||
return &tf, nil | ||
} | ||
|
||
type utilityControl struct { | ||
conf utilityControlConfig | ||
|
||
mu sync.Mutex | ||
agg aggregate.Aggregate | ||
} | ||
|
||
func (tf *utilityControl) Transform(_ context.Context, msg *message.Message) ([]*message.Message, error) { | ||
tf.mu.Lock() | ||
defer tf.mu.Unlock() | ||
|
||
if msg.IsControl() { | ||
// If a control message is received, then the aggregation is reset | ||
// to prevent sending duplicate control messages. | ||
tf.agg.ResetAll() | ||
|
||
return []*message.Message{msg}, nil | ||
} | ||
|
||
if ok := tf.agg.Add("", msg.Data()); ok { | ||
return []*message.Message{msg}, nil | ||
} | ||
|
||
tf.agg.Reset("") | ||
if ok := tf.agg.Add("", msg.Data()); !ok { | ||
return nil, fmt.Errorf("transform: utility_control: %v", errSendBatchMisconfigured) | ||
} | ||
|
||
ctrl := message.New().AsControl() | ||
return []*message.Message{msg, ctrl}, nil | ||
} | ||
|
||
func (tf *utilityControl) String() string { | ||
b, _ := json.Marshal(tf.conf) | ||
return string(b) | ||
} |