Merge pull request #163 from jwetzell/feat/pub-sub-module-type

convert nats and mqtt to a generic pub/sub module type

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
+}

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) {
+		New: func(moduleConfig config.ModuleConfig) (common.Module, error) {
-			params := config.Params
+			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 {
+func (mc *MQTTClient) Publish(ctx context.Context, topic string, payload any) error {
-	payloadMessage, ok := common.GetAnyAs[mqtt.Message](payload)
+	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()
 

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
 }

internal/processor/mqtt-message-create.go

@@ -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
-		},
-	})
-}

internal/processor/nats-message-create.go

@@ -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
-		},
-	})
-}

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
+		},
+	})
+}

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
 	}{}

internal/processor/test/mqtt-message-create_test.go

@@ -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++
-	}
-}

internal/processor/test/nats-message-create_test.go

@@ -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++
-	}
-}

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)
+			}
+		})
+	}
+}

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
+}