diff --git a/internal/common/module.go b/internal/common/module.go index 6517b14..47d133c 100644 --- a/internal/common/module.go +++ b/internal/common/module.go @@ -24,3 +24,7 @@ type KeyValueModule interface { type DatabaseModule interface { QueryContext(ctx context.Context, query string, args ...any) (*sql.Rows, error) } + +type PubSubModule interface { + Publish(ctx context.Context, topic string, payload any) error +} diff --git a/internal/module/mqtt-client.go b/internal/module/mqtt-client.go index ad5bcb4..2c56f63 100644 --- a/internal/module/mqtt-client.go +++ b/internal/module/mqtt-client.go @@ -2,6 +2,7 @@ package module import ( "context" + "encoding/json" "errors" "fmt" "log/slog" @@ -20,6 +21,8 @@ type MQTTClient struct { Broker string ClientID string Topic string + QoS byte + Retained bool client mqtt.Client logger *slog.Logger cancel context.CancelFunc @@ -45,12 +48,24 @@ func init() { Title: "Client ID", Type: "string", }, + "qos": { + Title: "QoS", + Type: "integer", + Minimum: jsonschema.Ptr[float64](0), + Maximum: jsonschema.Ptr[float64](2), + Default: json.RawMessage(`0`), + }, + "retained": { + Title: "Retained", + Type: "boolean", + Default: json.RawMessage(`false`), + }, }, Required: []string{"broker", "topic", "clientId"}, AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}}, }, - New: func(config config.ModuleConfig) (common.Module, error) { - params := config.Params + New: func(moduleConfig config.ModuleConfig) (common.Module, error) { + params := moduleConfig.Params brokerString, err := params.GetString("broker") if err != nil { @@ -69,7 +84,27 @@ func init() { return nil, fmt.Errorf("mqtt.client clientId error: %w", err) } - return &MQTTClient{config: config, Broker: brokerString, Topic: topicString, ClientID: clientIdString, logger: CreateLogger(config)}, nil + qosString, err := params.GetInt("qos") + + if err != nil { + if errors.Is(err, config.ErrParamNotFound) { + qosString = 0 + } else { + return nil, fmt.Errorf("mqtt.client qos error: %w", err) + } + } + + retainedBool, err := params.GetBool("retained") + + if err != nil { + if errors.Is(err, config.ErrParamNotFound) { + retainedBool = false + } else { + return nil, fmt.Errorf("mqtt.client retained error: %w", err) + } + } + + return &MQTTClient{config: moduleConfig, Broker: brokerString, Topic: topicString, ClientID: clientIdString, QoS: byte(qosString), Retained: retainedBool, logger: CreateLogger(moduleConfig)}, nil }, }) } @@ -118,11 +153,15 @@ func (mc *MQTTClient) Start(ctx context.Context, router common.RouteIO) error { return nil } -func (mc *MQTTClient) Output(ctx context.Context, payload any) error { - payloadMessage, ok := common.GetAnyAs[mqtt.Message](payload) +func (mc *MQTTClient) Publish(ctx context.Context, topic string, payload any) error { + payloadBytes, ok := common.GetAnyAsByteSlice(payload) if !ok { - return errors.New("mqtt.client is only able to output a MQTTMessage") + payloadString, ok := common.GetAnyAs[string](payload) + if !ok { + return errors.New("mqtt.client is only able to publish bytes or string") + } + payloadBytes = []byte(payloadString) } if mc.client == nil { @@ -133,7 +172,7 @@ func (mc *MQTTClient) Output(ctx context.Context, payload any) error { return errors.New("mqtt.client is not connected") } - token := mc.client.Publish(payloadMessage.Topic(), payloadMessage.Qos(), payloadMessage.Retained(), payloadMessage.Payload()) + token := mc.client.Publish(topic, mc.QoS, mc.Retained, payloadBytes) token.Wait() diff --git a/internal/module/nats-client.go b/internal/module/nats-client.go index 5724115..5f3a647 100644 --- a/internal/module/nats-client.go +++ b/internal/module/nats-client.go @@ -9,7 +9,6 @@ import ( "github.com/google/jsonschema-go/jsonschema" "github.com/jwetzell/showbridge-go/internal/common" "github.com/jwetzell/showbridge-go/internal/config" - "github.com/jwetzell/showbridge-go/internal/processor" "github.com/nats-io/nats.go" ) @@ -106,12 +105,16 @@ func (nc *NATSClient) Start(ctx context.Context, router common.RouteIO) error { return nil } -func (nc *NATSClient) Output(ctx context.Context, payload any) error { +func (nc *NATSClient) Publish(ctx context.Context, topic string, payload any) error { - payloadMessage, ok := common.GetAnyAs[processor.NATSMessage](payload) + payloadBytes, ok := common.GetAnyAsByteSlice(payload) if !ok { - return errors.New("nats.client is only able to output NATSMessage") + payloadString, ok := common.GetAnyAs[string](payload) + if !ok { + return errors.New("nats.client is only able to publish bytes or string") + } + payloadBytes = []byte(payloadString) } nc.clientMu.Lock() @@ -125,7 +128,7 @@ func (nc *NATSClient) Output(ctx context.Context, payload any) error { return errors.New("nats.client is not connected") } - err := nc.client.Publish(payloadMessage.Subject, payloadMessage.Payload) + err := nc.client.Publish(topic, payloadBytes) return err } diff --git a/internal/processor/mqtt-message-create.go b/internal/processor/mqtt-message-create.go deleted file mode 100644 index 0c43c5c..0000000 --- a/internal/processor/mqtt-message-create.go +++ /dev/null @@ -1,145 +0,0 @@ -package processor - -import ( - "context" - "errors" - "fmt" - - "github.com/google/jsonschema-go/jsonschema" - "github.com/jwetzell/showbridge-go/internal/common" - "github.com/jwetzell/showbridge-go/internal/config" -) - -type MQTTMessage struct { - topic string - qos byte - payload []byte - retained bool -} - -type MQTTMessageCreate struct { - config config.ProcessorConfig - Topic string - QoS byte - Retained bool - Payload []byte -} - -func NewMQTTMessage(topic string, qos byte, retained bool, payload []byte) MQTTMessage { - return MQTTMessage{ - topic: topic, - qos: qos, - retained: retained, - payload: payload, - } -} - -func (mm MQTTMessage) Duplicate() bool { - // TODO(jwetzell): implement? - return false -} - -func (mm MQTTMessage) Qos() byte { - return mm.qos -} - -func (mm MQTTMessage) Retained() bool { - return mm.retained -} - -func (mm MQTTMessage) Topic() string { - return mm.topic -} - -func (mm MQTTMessage) MessageID() uint16 { - // TODO(jwetzell): implement? - return 0 -} - -func (mm MQTTMessage) Payload() []byte { - return mm.payload -} - -func (mm MQTTMessage) Ack() {} - -func (mmc *MQTTMessageCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) { - // TODO(jwetzell): support templating - - wrappedPayload.Payload = MQTTMessage{ - topic: mmc.Topic, - qos: mmc.QoS, - retained: mmc.Retained, - payload: mmc.Payload, - } - - return wrappedPayload, nil -} - -func (mmc *MQTTMessageCreate) Type() string { - return mmc.config.Type -} - -func init() { - RegisterProcessor(ProcessorRegistration{ - Type: "mqtt.message.create", - Title: "Create MQTT Message", - ParamsSchema: &jsonschema.Schema{ - Type: "object", - Properties: map[string]*jsonschema.Schema{ - "topic": { - Title: "Topic", - Type: "string", - }, - "qos": { - Title: "QoS", - Type: "number", - }, - "retained": { - Title: "Retained", - Type: "boolean", - }, - "payload": { - Title: "Payload", - Type: "string", - }, - }, - Required: []string{"topic", "qos", "retained", "payload"}, - AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}}, - }, - New: func(processorConfig config.ProcessorConfig) (Processor, error) { - params := processorConfig.Params - topicString, err := params.GetString("topic") - if err != nil { - return nil, fmt.Errorf("mqtt.message.create topic error: %w", err) - } - - qosByte, err := params.GetInt("qos") - if err != nil { - return nil, fmt.Errorf("mqtt.message.create qos error: %w", err) - } - - retainedBool, err := params.GetBool("retained") - if err != nil { - return nil, fmt.Errorf("mqtt.message.create retained error: %w", err) - } - - //TODO(jwetzell): convert payload into []byte or string for sending - payloadString, err := params.GetString("payload") - if err != nil { - if errors.Is(err, config.ErrParamNotString) { - payloadBytes, err := params.GetByteSlice("payload") - if err != nil { - return nil, fmt.Errorf("mqtt.message.create payload error: %w", err) - } - return &MQTTMessageCreate{config: processorConfig, Topic: topicString, QoS: byte(qosByte), Retained: retainedBool, Payload: payloadBytes}, nil - } else { - return nil, fmt.Errorf("mqtt.message.create payload error: %w", err) - } - } - - payloadBytes := []byte(payloadString) - - return &MQTTMessageCreate{config: processorConfig, Topic: topicString, QoS: byte(qosByte), Retained: retainedBool, Payload: payloadBytes}, nil - }, - }) -} diff --git a/internal/processor/nats-message-create.go b/internal/processor/nats-message-create.go deleted file mode 100644 index 6531fb8..0000000 --- a/internal/processor/nats-message-create.go +++ /dev/null @@ -1,107 +0,0 @@ -package processor - -import ( - "bytes" - "context" - "fmt" - "text/template" - - "github.com/google/jsonschema-go/jsonschema" - "github.com/jwetzell/showbridge-go/internal/common" - "github.com/jwetzell/showbridge-go/internal/config" -) - -type NATSMessage struct { - Subject string - Payload []byte -} - -type NATSMessageCreate struct { - config config.ProcessorConfig - Subject *template.Template - Payload *template.Template -} - -func (nmc *NATSMessageCreate) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) { - - templateData := wrappedPayload - - var payloadBuffer bytes.Buffer - err := nmc.Payload.Execute(&payloadBuffer, templateData) - - if err != nil { - wrappedPayload.End = true - return wrappedPayload, err - } - - payloadString := payloadBuffer.String() - - var subjectBuffer bytes.Buffer - err = nmc.Subject.Execute(&subjectBuffer, templateData) - - if err != nil { - wrappedPayload.End = true - return wrappedPayload, err - } - - subjectString := subjectBuffer.String() - - wrappedPayload.Payload = NATSMessage{ - Subject: subjectString, - Payload: []byte(payloadString), - } - - return wrappedPayload, nil -} - -func (nmc *NATSMessageCreate) Type() string { - return nmc.config.Type -} - -func init() { - RegisterProcessor(ProcessorRegistration{ - Type: "nats.message.create", - Title: "Create NATS Message", - ParamsSchema: &jsonschema.Schema{ - Type: "object", - Properties: map[string]*jsonschema.Schema{ - "subject": { - Title: "Subject", - Type: "string", - }, - "payload": { - Title: "Payload", - Type: "string", - }, - }, - Required: []string{"subject", "payload"}, - AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}}, - }, - New: func(config config.ProcessorConfig) (Processor, error) { - params := config.Params - subjectString, err := params.GetString("subject") - if err != nil { - return nil, fmt.Errorf("nats.message.create subject error: %w", err) - } - - subjectTemplate, err := template.New("subject").Parse(subjectString) - - if err != nil { - return nil, err - } - - payloadString, err := params.GetString("payload") - if err != nil { - return nil, fmt.Errorf("nats.message.create payload error: %w", err) - } - - payloadTemplate, err := template.New("payload").Parse(payloadString) - - if err != nil { - return nil, err - } - - return &NATSMessageCreate{config: config, Subject: subjectTemplate, Payload: payloadTemplate}, nil - }, - }) -} diff --git a/internal/processor/pubsub-publish.go b/internal/processor/pubsub-publish.go new file mode 100644 index 0000000..b9c7a4e --- /dev/null +++ b/internal/processor/pubsub-publish.go @@ -0,0 +1,109 @@ +package processor + +import ( + "bytes" + "context" + "errors" + "fmt" + "log/slog" + "text/template" + + "github.com/google/jsonschema-go/jsonschema" + "github.com/jwetzell/showbridge-go/internal/common" + "github.com/jwetzell/showbridge-go/internal/config" +) + +type PubSubPublish struct { + config config.ProcessorConfig + ModuleId string + Topic *template.Template + logger *slog.Logger + module common.PubSubModule +} + +func (psp *PubSubPublish) Process(ctx context.Context, wrappedPayload common.WrappedPayload) (common.WrappedPayload, error) { + if psp.module == nil { + if wrappedPayload.Modules == nil { + wrappedPayload.End = true + return wrappedPayload, errors.New("pubsub.publish wrapped payload has no modules") + } + + module, ok := wrappedPayload.Modules[psp.ModuleId] + if !ok { + wrappedPayload.End = true + return wrappedPayload, fmt.Errorf("pubsub.publish unable to find module with id: %s", psp.ModuleId) + } + + dbModule, ok := module.(common.PubSubModule) + if !ok { + wrappedPayload.End = true + return wrappedPayload, fmt.Errorf("pubsub.publish module with id %s is not an OutputModule", psp.ModuleId) + } + psp.module = dbModule + } + + var topicBuffer bytes.Buffer + err := psp.Topic.Execute(&topicBuffer, wrappedPayload) + + if err != nil { + wrappedPayload.End = true + return wrappedPayload, err + } + + err = psp.module.Publish(ctx, topicBuffer.String(), wrappedPayload.Payload) + if err != nil { + wrappedPayload.End = true + return wrappedPayload, fmt.Errorf("pubsub.publish error publishing: %w", err) + } + + return wrappedPayload, nil +} + +func (psp *PubSubPublish) Type() string { + return psp.config.Type +} + +func init() { + RegisterProcessor(ProcessorRegistration{ + Type: "pubsub.publish", + Title: "Publish to Pub/Sub Topic", + ParamsSchema: &jsonschema.Schema{ + Type: "object", + Properties: map[string]*jsonschema.Schema{ + "module": { + Title: "Module ID", + Type: "string", + Description: "ID of the module to publish to", + }, + "topic": { + Title: "Topic", + Type: "string", + Description: "Topic to publish to", + }, + }, + Required: []string{"module", "topic"}, + AdditionalProperties: &jsonschema.Schema{Not: &jsonschema.Schema{}}, + }, + New: func(config config.ProcessorConfig) (Processor, error) { + + params := config.Params + + moduleIdString, err := params.GetString("module") + if err != nil { + return nil, fmt.Errorf("pubsub.publish module error: %w", err) + } + + topicString, err := params.GetString("topic") + if err != nil { + return nil, fmt.Errorf("pubsub.publish topic error: %w", err) + } + + topicTemplate, err := template.New("topic").Parse(topicString) + + if err != nil { + return nil, err + } + return &PubSubPublish{config: config, ModuleId: moduleIdString, Topic: topicTemplate, logger: slog.Default().With("component", "processor", "type", config.Type)}, nil + }, + }) +} diff --git a/internal/processor/test/http-request-do_test.go b/internal/processor/test/http-request-do_test.go index db03ff2..c56b732 100644 --- a/internal/processor/test/http-request-do_test.go +++ b/internal/processor/test/http-request-do_test.go @@ -36,7 +36,7 @@ func TestGoodHTTPRequestDo(t *testing.T) { tests := []struct { name string - expected processor.NATSMessage + expected any params map[string]any payload any }{} diff --git a/internal/processor/test/mqtt-message-create_test.go b/internal/processor/test/mqtt-message-create_test.go deleted file mode 100644 index efffb56..0000000 --- a/internal/processor/test/mqtt-message-create_test.go +++ /dev/null @@ -1,272 +0,0 @@ -package processor_test - -import ( - "reflect" - "testing" - - mqtt "github.com/eclipse/paho.mqtt.golang" - "github.com/jwetzell/showbridge-go/internal/common" - "github.com/jwetzell/showbridge-go/internal/config" - "github.com/jwetzell/showbridge-go/internal/processor" -) - -func TestMQTTMessageCreateFromRegistry(t *testing.T) { - registration, ok := processor.ProcessorRegistry["mqtt.message.create"] - if !ok { - t.Fatalf("mqtt.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "mqtt.message.create", - Params: map[string]any{ - "topic": "test/topic", - "payload": "Hello, World!", - "qos": 1, - "retained": true, - }, - }) - - if err != nil { - t.Fatalf("failed to create mqtt.message.create processor: %s", err) - } - - if processorInstance.Type() != "mqtt.message.create" { - t.Fatalf("mqtt.message.create processor has wrong type: %s", processorInstance.Type()) - } -} - -func TestGoodMQTTMessageCreate(t *testing.T) { - tests := []struct { - name string - payload any - params map[string]any - expected any - }{ - { - name: "basic topic and string payload", - params: map[string]any{ - "topic": "test/topic", - "payload": "Hello, World!", - "qos": 1, - "retained": true, - }, - payload: "test", - expected: processor.NewMQTTMessage("test/topic", 1, true, []byte("Hello, World!")), - }, - { - name: "basic topic and []byte payload", - params: map[string]any{ - "topic": "test/topic", - "payload": []byte{72, 101, 108, 108, 111}, - "qos": 1, - "retained": true, - }, - payload: "test", - expected: processor.NewMQTTMessage("test/topic", 1, true, []byte("Hello")), - }, - { - name: "basic topic and []int payload", - params: map[string]any{ - "topic": "test/topic", - "payload": []int{72, 101, 108, 108, 111}, - "qos": 1, - "retained": true, - }, - payload: "test", - expected: processor.NewMQTTMessage("test/topic", 1, true, []byte("Hello")), - }, - { - name: "basic topic and []uint payload", - params: map[string]any{ - "topic": "test/topic", - "payload": []uint{72, 101, 108, 108, 111}, - "qos": 1, - "retained": true, - }, - payload: "test", - expected: processor.NewMQTTMessage("test/topic", 1, true, []byte("Hello")), - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - registration, ok := processor.ProcessorRegistry["mqtt.message.create"] - if !ok { - t.Fatalf("mqtt.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "mqtt.message.create", - Params: test.params, - }) - - if err != nil { - t.Fatalf("mqtt.message.create failed to create processor: %s", err) - } - - got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload}) - - if err != nil { - t.Fatalf("mqtt.message.create processing failed: %s", err) - } - - if test.expected == nil { - if got.Payload != nil { - t.Fatalf("mqtt.message.create got %+v, expected nil", got) - } - return - } - - gotMessage, ok := got.Payload.(mqtt.Message) - if !ok { - t.Fatalf("mqtt.message.create returned a %T payload: %+v", got, got) - } - - if !reflect.DeepEqual(gotMessage, test.expected) { - t.Fatalf("mqtt.message.create got %+v, expected %+v", gotMessage, test.expected) - } - }) - } -} - -func TestBadMQTTMessageCreate(t *testing.T) { - tests := []struct { - name string - params map[string]any - payload any - errorString string - }{ - { - name: "no topic parameter", - params: map[string]any{}, - payload: "test", - errorString: "mqtt.message.create topic error: not found", - }, - { - name: "non-string topic parameter", - params: map[string]any{ - "topic": 1, - }, - payload: "test", - errorString: "mqtt.message.create topic error: not a string", - }, - { - name: "no qos parameter", - params: map[string]any{ - "topic": "test/topic", - }, - payload: "test", - errorString: "mqtt.message.create qos error: not found", - }, - { - name: "non-number qos parameter", - params: map[string]any{ - "topic": "test/topic", - "qos": "1", - }, - payload: "test", - errorString: "mqtt.message.create qos error: not a number", - }, - { - name: "no retained parameter", - params: map[string]any{ - "topic": "test/topic", - "qos": 1, - }, - payload: "test", - errorString: "mqtt.message.create retained error: not found", - }, - { - name: "non-bool retained parameter", - params: map[string]any{ - "topic": "test/topic", - "qos": 1, - "retained": "1", - }, - payload: "test", - errorString: "mqtt.message.create retained error: not a boolean", - }, - { - name: "no payload parameter", - params: map[string]any{ - "topic": "test/topic", - "qos": 1, - "retained": true, - }, - payload: "test", - errorString: "mqtt.message.create payload error: not found", - }, - { - name: "non-string payload parameter", - params: map[string]any{ - "topic": "test/topic", - "qos": 1, - "retained": true, - "payload": 123, - }, - payload: 1, - errorString: "mqtt.message.create payload error: not a byte slice", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - registration, ok := processor.ProcessorRegistry["mqtt.message.create"] - if !ok { - t.Fatalf("mqtt.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "mqtt.message.create", - Params: test.params, - }) - - if err != nil { - if test.errorString != err.Error() { - t.Fatalf("mqtt.message.create got error '%s', expected '%s'", err.Error(), test.errorString) - } - return - } - - got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload}) - - if err == nil { - t.Fatalf("mqtt.message.create expected to fail but succeeded, got: %v", got) - } - - if err.Error() != test.errorString { - t.Fatalf("mqtt.message.create got error '%s', expected '%s'", err.Error(), test.errorString) - } - }) - } -} - -func BenchmarkMQTTMessageCreate(b *testing.B) { - registration, ok := processor.ProcessorRegistry["mqtt.message.create"] - if !ok { - b.Fatalf("mqtt.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "mqtt.message.create", - Params: map[string]any{ - "topic": "test/topic", - "qos": 1, - "retained": true, - "payload": "{{.Payload}}", - }, - }) - - if err != nil { - b.Fatalf("mqtt.message.create failed to create processor: %s", err) - } - - count := 0 - for b.Loop() { - _, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count}) - if err != nil { - b.Fatalf("mqtt.message.create processing failed: %s", err) - } - count++ - } -} diff --git a/internal/processor/test/nats-message-create_test.go b/internal/processor/test/nats-message-create_test.go deleted file mode 100644 index c77725a..0000000 --- a/internal/processor/test/nats-message-create_test.go +++ /dev/null @@ -1,238 +0,0 @@ -package processor_test - -import ( - "reflect" - "testing" - - "github.com/jwetzell/showbridge-go/internal/common" - "github.com/jwetzell/showbridge-go/internal/config" - "github.com/jwetzell/showbridge-go/internal/processor" -) - -func TestNATSMessageCreateFromRegistry(t *testing.T) { - registration, ok := processor.ProcessorRegistry["nats.message.create"] - if !ok { - t.Fatalf("nats.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "nats.message.create", - Params: map[string]any{ - "subject": "test", - "payload": "Hello, World!", - }, - }) - - if err != nil { - t.Fatalf("failed to create nats.message.create processor: %s", err) - } - - if processorInstance.Type() != "nats.message.create" { - t.Fatalf("nats.message.create processor has wrong type: %s", processorInstance.Type()) - } -} - -func TestGoodNATSMessageCreate(t *testing.T) { - - tests := []struct { - name string - expected processor.NATSMessage - params map[string]any - payload any - }{ - { - name: "simple payload", - params: map[string]any{ - "subject": "test", - "payload": "Hello, World!", - }, - payload: nil, - expected: processor.NATSMessage{ - Subject: "test", - Payload: []byte("Hello, World!"), - }, - }, - { - name: "payload with template", - params: map[string]any{ - "subject": "test", - "payload": "Hello, {{.Payload.Name}}!", - }, - payload: map[string]any{ - "Name": "Alice", - }, - expected: processor.NATSMessage{ - Subject: "test", - Payload: []byte("Hello, Alice!"), - }, - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - - registration, ok := processor.ProcessorRegistry["nats.message.create"] - if !ok { - t.Fatalf("nats.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "nats.message.create", - Params: test.params, - }) - - if err != nil { - t.Fatalf("nats.message.create failed to create processor: %s", err) - } - - got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload}) - if err != nil { - t.Fatalf("nats.message.create processing failed: %s", err) - } - - if !reflect.DeepEqual(got.Payload, test.expected) { - t.Fatalf("nats.message.create got %+v (%T), expected %+v (%T)", got.Payload, got.Payload, test.expected, test.expected) - } - }) - } -} - -func TestBadNATSMessageCreate(t *testing.T) { - tests := []struct { - name string - params map[string]any - payload any - errorString string - }{ - { - name: "missing subject param", - params: map[string]any{ - "payload": "Hello, World!", - }, - payload: nil, - errorString: "nats.message.create subject error: not found", - }, - { - name: "subject param not a string", - params: map[string]any{ - "subject": 123, - "payload": "Hello, World!", - }, - payload: nil, - errorString: "nats.message.create subject error: not a string", - }, - { - name: "missing payload param", - params: map[string]any{ - "subject": "test", - }, - payload: nil, - errorString: "nats.message.create payload error: not found", - }, - { - name: "payload param not a string", - params: map[string]any{ - "subject": "test", - "payload": 123, - }, - payload: nil, - errorString: "nats.message.create payload error: not a string", - }, - { - name: "payload template error", - params: map[string]any{ - "subject": "test", - "payload": "Hello, {{.Payload.Name}}!", - }, - payload: nil, - errorString: "template: payload:1:17: executing \"payload\" at <.Payload.Name>: nil pointer evaluating interface {}.Name", - }, - { - name: "subject template error", - params: map[string]any{ - "subject": "test.{{.Payload.Name}}", - "payload": "Hello, World!", - }, - payload: nil, - errorString: "template: subject:1:15: executing \"subject\" at <.Payload.Name>: nil pointer evaluating interface {}.Name", - }, - { - name: "subject template syntax error", - params: map[string]any{ - "subject": "{{.Payload.Name", - "payload": "Hello, World!", - }, - payload: nil, - errorString: "template: subject:1: unclosed action", - }, - { - name: "payload template syntax error", - params: map[string]any{ - "subject": "test", - "payload": "Hello, {{.Payload.Name", - }, - payload: nil, - errorString: "template: payload:1: unclosed action", - }, - } - - for _, test := range tests { - t.Run(test.name, func(t *testing.T) { - - registration, ok := processor.ProcessorRegistry["nats.message.create"] - if !ok { - t.Fatalf("nats.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "nats.message.create", - Params: test.params, - }) - - if err != nil { - if test.errorString != err.Error() { - t.Fatalf("nats.message.create got error '%s', expected '%s'", err.Error(), test.errorString) - } - return - } - - got, err := processorInstance.Process(t.Context(), common.WrappedPayload{Payload: test.payload}) - - if err == nil { - t.Fatalf("nats.message.create expected to fail but succeeded, got: %v", got) - } - - if err.Error() != test.errorString { - t.Fatalf("nats.message.create got error '%s', expected '%s'", err.Error(), test.errorString) - } - }) - } -} - -func BenchmarkNATSMessageCreate(b *testing.B) { - registration, ok := processor.ProcessorRegistry["nats.message.create"] - if !ok { - b.Fatalf("nats.message.create processor not registered") - } - - processorInstance, err := registration.New(config.ProcessorConfig{ - Type: "nats.message.create", - Params: map[string]any{ - "subject": "test.subject", - "payload": "{{.Payload}}", - }, - }) - - if err != nil { - b.Fatalf("nats.message.create failed to create processor: %s", err) - } - - count := 0 - for b.Loop() { - _, err := processorInstance.Process(b.Context(), common.WrappedPayload{Payload: count}) - if err != nil { - b.Fatalf("nats.message.create processing failed: %s", err) - } - count++ - } -} diff --git a/internal/processor/test/pubsub-publish_test.go b/internal/processor/test/pubsub-publish_test.go new file mode 100644 index 0000000..c2a67cb --- /dev/null +++ b/internal/processor/test/pubsub-publish_test.go @@ -0,0 +1,251 @@ +package processor_test + +import ( + "reflect" + "testing" + + "github.com/jwetzell/showbridge-go/internal/common" + "github.com/jwetzell/showbridge-go/internal/config" + "github.com/jwetzell/showbridge-go/internal/processor" + "github.com/jwetzell/showbridge-go/internal/test" + _ "modernc.org/sqlite" +) + +func TestPubSubPublishFromRegistry(t *testing.T) { + registration, ok := processor.ProcessorRegistry["pubsub.publish"] + if !ok { + t.Fatalf("pubsub.publish processor not registered") + } + + processorInstance, err := registration.New(config.ProcessorConfig{ + Type: "pubsub.publish", + Params: map[string]any{ + "module": "test", + "topic": "test", + }, + }) + if err != nil { + t.Fatalf("failed to create pubsub.publish processor: %s", err) + } + + if processorInstance.Type() != "pubsub.publish" { + t.Fatalf("pubsub.publish processor has wrong type: %s", processorInstance.Type()) + } + + payload := "hello" + expected := "hello" + + got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ + Payload: payload, + Modules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + }) + if err != nil { + t.Fatalf("pubsub.publish processing failed: %s", err) + } + + if !reflect.DeepEqual(got.Payload, expected) { + t.Fatalf("pubsub.publish got %+v, expected %+v", got.Payload, expected) + } +} + +func TestGoodPubSubPublish(t *testing.T) { + + testCases := []struct { + name string + params map[string]any + payload any + expected any + }{ + { + name: "basic topic", + params: map[string]any{ + "module": "test", + "topic": "test", + }, + payload: "", + expected: "", + }, + } + for _, testCase := range testCases { + t.Run(testCase.name, func(t *testing.T) { + registration, ok := processor.ProcessorRegistry["pubsub.publish"] + if !ok { + t.Fatalf("pubsub.publish processor not registered") + } + + processorInstance, err := registration.New(config.ProcessorConfig{ + Type: "pubsub.publish", + Params: testCase.params, + }) + + if err != nil { + t.Fatalf("pubsub.publish failed to create processor: %s", err) + } + + got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ + Modules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + Payload: testCase.payload, + }) + + if err != nil { + t.Fatalf("pubsub.publish processing failed: %s", err) + } + + if !reflect.DeepEqual(got.Payload, testCase.expected) { + t.Fatalf("pubsub.publish got payload: %+v, expected %+v", got.Payload, testCase.expected) + } + }) + } +} + +func TestBadPubSubPublish(t *testing.T) { + tests := []struct { + name string + params map[string]any + payload any + wrappedPayloadModules map[string]common.Module + errorString string + }{ + { + name: "no module param", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "topic": "test", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "pubsub.publish module error: not found", + }, + { + name: "non string module", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": 1, + "topic": "test", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "pubsub.publish module error: not a string", + }, + { + name: "no topic param", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "pubsub.publish topic error: not found", + }, + { + name: "non string topic", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": 1, + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "pubsub.publish topic error: not a string", + }, + { + name: "topic template syntax error", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": "{{", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "template: topic:1: unclosed action", + }, + { + name: "topic template error", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": "{{.Data}}", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestPubSubModule("test"), + }, + errorString: "template: topic:1:2: executing \"topic\" at <.Data>: can't evaluate field Data in type common.WrappedPayload", + }, + { + name: "no modules in context", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": "test", + }, + wrappedPayloadModules: nil, + errorString: "pubsub.publish wrapped payload has no modules", + }, + { + name: "module not found in context", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": "test", + }, + wrappedPayloadModules: map[string]common.Module{}, + errorString: "pubsub.publish unable to find module with id: test", + }, + { + name: "module not an OutputModule", + payload: test.TestStruct{Data: "hello"}, + params: map[string]any{ + "module": "test", + "topic": "test", + }, + wrappedPayloadModules: map[string]common.Module{ + "test": test.NewTestKVModule("test"), + }, + errorString: "pubsub.publish module with id test is not an OutputModule", + }, + } + + for _, test := range tests { + t.Run(test.name, func(t *testing.T) { + + registration, ok := processor.ProcessorRegistry["pubsub.publish"] + if !ok { + t.Fatalf("pubsub.publish processor not registered") + } + + processorInstance, err := registration.New(config.ProcessorConfig{ + Type: "pubsub.publish", + Params: test.params, + }) + + if err != nil { + if test.errorString != err.Error() { + t.Fatalf("pubsub.publish got error '%s', expected '%s'", err.Error(), test.errorString) + } + return + } + + got, err := processorInstance.Process(t.Context(), common.WrappedPayload{ + Payload: test.payload, + Modules: test.wrappedPayloadModules, + }) + + if err == nil { + t.Fatalf("pubsub.publish expected to fail but got payload: %+v", got) + } + + if err.Error() != test.errorString { + t.Fatalf("pubsub.publish got error '%s', expected '%s'", err.Error(), test.errorString) + } + }) + } +} diff --git a/internal/test/module.go b/internal/test/module.go index 5e55883..1aee83b 100644 --- a/internal/test/module.go +++ b/internal/test/module.go @@ -111,3 +111,32 @@ func (m *TestDBModule) Type() string { func (m *TestDBModule) Id() string { return m.id } + +func NewTestPubSubModule(id string) *TestPubSubModule { + return &TestPubSubModule{ + id: id, + } +} + +type TestPubSubModule struct { + id string +} + +func (m *TestPubSubModule) Start(ctx context.Context, router common.RouteIO) error { + <-ctx.Done() + return nil +} + +func (m *TestPubSubModule) Publish(ctx context.Context, topic string, payload any) error { + return nil +} + +func (m *TestPubSubModule) Stop() {} + +func (m *TestPubSubModule) Type() string { + return "test.pubsub" +} + +func (m *TestPubSubModule) Id() string { + return m.id +}