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

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