Merge pull request #167 from jwetzell/feat/kv-set-from-payload

switch kv.set to take value from incoming payload
This commit is contained in:
Joel Wetzell
2026-05-20 20:40:48 -05:00
committed by GitHub
2 changed files with 7 additions and 93 deletions
+3 -28
View File
@@ -1,11 +1,9 @@
package processor package processor
import ( import (
"bytes"
"context" "context"
"errors" "errors"
"fmt" "fmt"
"html/template"
"log/slog" "log/slog"
"github.com/google/jsonschema-go/jsonschema" "github.com/google/jsonschema-go/jsonschema"
@@ -17,7 +15,6 @@ type KVSet struct {
config config.ProcessorConfig config config.ProcessorConfig
ModuleId string ModuleId string
Key string Key string
Value *template.Template
logger *slog.Logger logger *slog.Logger
module common.KeyValueModule module common.KeyValueModule
} }
@@ -43,15 +40,7 @@ func (kvs *KVSet) Process(ctx context.Context, wrappedPayload common.WrappedPayl
kvs.module = kvModule kvs.module = kvModule
} }
var valueBuffer bytes.Buffer err := kvs.module.Set(kvs.Key, wrappedPayload.Payload)
err := kvs.Value.Execute(&valueBuffer, wrappedPayload)
if err != nil {
wrappedPayload.End = true
return wrappedPayload, err
}
err = kvs.module.Set(kvs.Key, valueBuffer.String())
if err != nil { if err != nil {
wrappedPayload.End = true wrappedPayload.End = true
return wrappedPayload, fmt.Errorf("kv.set error setting key: %w", err) return wrappedPayload, fmt.Errorf("kv.set error setting key: %w", err)
@@ -79,12 +68,8 @@ func init() {
Title: "Key", Title: "Key",
Type: "string", Type: "string",
}, },
"value": {
Title: "Value",
Type: "string",
},
}, },
Required: []string{"module", "key", "value"}, Required: []string{"module", "key"},
AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}}, AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}},
}, },
New: func(config config.ProcessorConfig) (Processor, error) { New: func(config config.ProcessorConfig) (Processor, error) {
@@ -101,17 +86,7 @@ func init() {
return nil, fmt.Errorf("kv.set key error: %w", err) return nil, fmt.Errorf("kv.set key error: %w", err)
} }
valueString, err := params.GetString("value") return &KVSet{config: config, ModuleId: moduleIdString, Key: keyString, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
if err != nil {
return nil, fmt.Errorf("kv.set value error: %w", err)
}
valueTemplate, err := template.New("template").Parse(valueString)
if err != nil {
return nil, err
}
return &KVSet{config: config, ModuleId: moduleIdString, Key: keyString, Value: valueTemplate, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil
}, },
}) })
} }
+4 -65
View File
@@ -21,7 +21,6 @@ func TestKvSetFromRegistry(t *testing.T) {
Params: map[string]any{ Params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "hello",
}, },
}) })
if err != nil { if err != nil {
@@ -32,8 +31,8 @@ func TestKvSetFromRegistry(t *testing.T) {
t.Fatalf("kv.set processor has wrong type: %s", processorInstance.Type()) t.Fatalf("kv.set processor has wrong type: %s", processorInstance.Type())
} }
payload := "" payload := "test"
expected := "" expected := "test"
got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ got, err := processorInstance.Process(t.Context(), common.WrappedPayload{
Modules: map[string]common.Module{ Modules: map[string]common.Module{
@@ -63,10 +62,9 @@ func TestGoodKvSet(t *testing.T) {
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "hello",
}, },
payload: "", payload: "test",
expected: "", expected: "test",
}, },
} }
for _, testCase := range testCases { for _, testCase := range testCases {
@@ -116,7 +114,6 @@ func TestBadKvSet(t *testing.T) {
payload: test.TestStruct{Data: "hello"}, payload: test.TestStruct{Data: "hello"},
params: map[string]any{ params: map[string]any{
"key": "test", "key": "test",
"value": "test",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{}, "test": &test.TestKVModule{},
@@ -129,7 +126,6 @@ func TestBadKvSet(t *testing.T) {
params: map[string]any{ params: map[string]any{
"module": 1, "module": 1,
"key": "test", "key": "test",
"value": "test",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{}, "test": &test.TestKVModule{},
@@ -141,7 +137,6 @@ func TestBadKvSet(t *testing.T) {
payload: test.TestStruct{Data: "hello"}, payload: test.TestStruct{Data: "hello"},
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"value": "test",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{}, "test": &test.TestKVModule{},
@@ -154,82 +149,28 @@ func TestBadKvSet(t *testing.T) {
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"key": 1, "key": 1,
"value": "test",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{}, "test": &test.TestKVModule{},
}, },
errorString: "kv.set key error: not a string", errorString: "kv.set key error: not a string",
}, },
{
name: "no value param",
payload: test.TestStruct{Data: "hello"},
params: map[string]any{
"module": "test",
"key": "test",
},
wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{},
},
errorString: "kv.set value error: not found",
},
{
name: "non string value",
payload: test.TestStruct{Data: "hello"},
params: map[string]any{
"module": "test",
"key": "test",
"value": 1,
},
wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{},
},
errorString: "kv.set value error: not a string",
},
{ {
name: "no modules in context", name: "no modules in context",
payload: test.TestStruct{Data: "hello"}, payload: test.TestStruct{Data: "hello"},
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "hello",
}, },
wrappedPayloadModules: nil, wrappedPayloadModules: nil,
errorString: "kv.set wrapped payload has no modules", errorString: "kv.set wrapped payload has no modules",
}, },
{
name: "value template syntax error",
payload: test.TestStruct{Data: "hello"},
params: map[string]any{
"module": "test",
"key": "test",
"value": "{{",
},
wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{},
},
errorString: "template: template:1: unclosed action",
},
{
name: "value template execution error",
payload: test.TestStruct{Data: "hello"},
params: map[string]any{
"module": "test",
"key": "test",
"value": "{{.Data}}",
},
wrappedPayloadModules: map[string]common.Module{
"test": &test.TestKVModule{},
},
errorString: "template: template:1:2: executing \"template\" at <.Data>: can't evaluate field Data in type common.WrappedPayload",
},
{ {
name: "module not found in context", name: "module not found in context",
payload: test.TestStruct{Data: "hello"}, payload: test.TestStruct{Data: "hello"},
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "hello",
}, },
wrappedPayloadModules: map[string]common.Module{}, wrappedPayloadModules: map[string]common.Module{},
errorString: "kv.set unable to find module with id: test", errorString: "kv.set unable to find module with id: test",
@@ -240,7 +181,6 @@ func TestBadKvSet(t *testing.T) {
params: map[string]any{ params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "hello",
}, },
wrappedPayloadModules: map[string]common.Module{ wrappedPayloadModules: map[string]common.Module{
"test": test.NewTestDBModule("test"), "test": test.NewTestDBModule("test"),
@@ -293,7 +233,6 @@ func BenchmarkKvSet(b *testing.B) {
Params: map[string]any{ Params: map[string]any{
"module": "test", "module": "test",
"key": "test", "key": "test",
"value": "{{.Payload}}",
}, },
}) })