Update configuration to handle starting broker and fail over

.gitignore

@@ -28,4 +28,5 @@ station/station
 otto.log
 debug*
 *.code-workspace
-*.log
+*.log
+coverage.*

config.go

@@ -0,0 +1,7 @@
+package otto
+
+import "github.com/rustyeddy/otto/messanger"
+
+type Config struct {
+	messanger.Config
+}

data/data.go

@@ -16,8 +16,8 @@ type Data interface {
 // Data is an array of timestamps and values representing the same
 // source of data over a period of time
 type DataPoint struct {
-	value     any       `json:"value"`
-	timestamp time.Time `json:"time-increment"`
+	value     any
+	timestamp time.Time
 }
 
 func NewData(dat any, ts time.Time) Data {

messanger/messanger.go

@@ -34,25 +34,26 @@ import (
 	"fmt"
 	"log/slog"
 	"net/http"
+	"os"
+	"strings"
 	"sync"
 )
 
-var (
-	// messanger holds the singleton instance of the active Messanger
-	messanger     Messanger
-	messangerLock sync.Mutex
-)
-
 // MessangerConfig holds configuration parameters for messanger initialization.
 // Currently supports broker address configuration for MQTT-based messangers.
-type MessangerConfig struct {
+type Config struct {
 	// Broker is the address of the MQTT broker (hostname or IP)
-	Broker string
+	Broker   string `json:"broker"`
+	Username string `json:"username"`
+	Password string `json:"password"`
 }
 
-var messangerConfig = MessangerConfig{
-	Broker: "localhost",
-}
+var (
+	// messanger holds the singleton instance of the active Messanger
+	msgr          Messanger
+	messangerLock sync.Mutex
+	config        Config
+)
 
 // MsgHandler is a callback function type for handling incoming messages.
 // Subscribers provide a MsgHandler function that will be invoked when
@@ -112,9 +113,9 @@ type Messanger interface {
 //
 // Supported ID values:
 //   - "none": Creates a local in-process messanger without MQTT
-//   - "mqtt": Creates an MQTT messanger connecting to an external broker
 //   - "local": Starts an embedded MQTT broker and creates an MQTT messanger
-//
+//   - default: Creates an MQTT messanger connecting to an external broker
+
 // The created messanger becomes the global singleton accessible via GetMessanger().
 // If an invalid ID is provided, logs an error and returns nil.
 //
@@ -124,26 +125,24 @@ type Messanger interface {
 //	if msg == nil {
 //	    log.Fatal("Failed to create messanger")
 //	}
-func NewMessanger(id string) (m Messanger) {
-	switch id {
+func NewMessanger(broker string) (m Messanger) {
+
+	switch broker {
 	case "none":
-		m = NewMessangerLocal(id)
-	case "mqtt":
-		m = NewMessangerMQTT(id, messangerConfig.Broker)
-	case "local":
+		msgr = NewMessangerLocal(broker)
+
+	case "otto":
 		_, err := StartMQTTBroker(context.Background())
 		if err != nil {
 			slog.Error("Failed to start embedded MQTT broker", "error", err)
 			return nil
 		}
-		m = NewMessangerMQTT(id, messangerConfig.Broker)
+		msgr = NewMessangerMQTT(broker, broker)
 
 	default:
-		slog.Error("Unknown messanger ID", "id", id)
-		return nil
+		msgr = NewMessangerMQTT(broker, broker)
 	}
-	messanger = m
-	return m
+	return msgr
 }
 
 // GetMessanger returns the singleton Messanger instance created by NewMessanger.
@@ -159,7 +158,60 @@ func NewMessanger(id string) (m Messanger) {
 func GetMessanger() Messanger {
 	messangerLock.Lock()
 	defer messangerLock.Unlock()
-	return messanger
+
+	if msgr != nil {
+		return msgr
+	}
+
+	// Take broker from the config first
+	broker := config.Broker
+	if broker == "" {
+		// if no config look for environment variable
+		broker = os.Getenv("MQTT_BROKER")
+	}
+	if broker == "" {
+		// if no environment variable then default to the built in
+		// broker
+		broker = "otto"
+	}
+
+	user := config.Username
+	if user == "" {
+		user = os.Getenv("MQTT_USERNAME")
+	}
+	pass := config.Password
+	if pass == "" {
+		pass = os.Getenv("MQTT_PASSWORD")
+	}
+
+	switch broker {
+	case "none":
+		msgr = NewMessangerLocal(broker)
+
+	case "otto":
+		var err error
+		shutdown, err = StartMQTTBroker(context.Background())
+		if err != nil {
+			slog.Error("Failed to start embedded MQTT broker", "error", err)
+
+			// A hack if bind address is in use, skip out and just
+			// use the client to bind to the already running broker
+			if !strings.Contains(err.Error(), "bind: address already in use") {
+				return nil
+			}
+
+			slog.Info("Assuming broker is already running, connecting to existing broker")
+		}
+		fallthrough
+
+	default:
+		msgr = NewMessangerMQTTWithAuth("otto", broker, user, pass)
+	}
+
+	ms := GetMsgSaver()
+	ms.Saving = true
+
+	return msgr
 }
 
 // MessangerBase provides a base implementation of the Messanger interface
@@ -246,6 +298,17 @@ func (mb *MessangerBase) PubMsg(msg *Msg) error {
 	return nil
 }
 
+func (mb *MessangerBase) Pub(topic string, data any) error {
+
+	b, err := Bytes(data)
+	if err != nil {
+		slog.Error("messanger failed to convert bytes", "error", err)
+		return err
+	}
+	msg := NewMsg(topic, b, "otto")
+	return mb.PubMsg(msg)
+}
+
 // Close cleanly shuts down the messanger.
 // This base implementation is a no-op that just logs the close operation.
 //

messanger/messanger_local.go

@@ -182,31 +182,3 @@ func (m *MessangerLocal) Close() {
 	}
 	slog.Debug("MessangerLocal.Close", "id", m.ID())
 }
-
-// toBytesUnchecked is an internal helper that converts common types to []byte.
-// It attempts to use the package's Bytes() function first, falling back to
-// basic type assertions if that fails.
-//
-// This is a best-effort conversion used internally by the local messanger.
-//
-// Supported types: []byte, string, and any other type (converted via fmt.Sprintf)
-//
-// Parameters:
-//   - v: The value to convert to bytes
-//
-// Returns the byte representation of the value.
-func toBytesUnchecked(v any) []byte {
-	// try using Bytes helper if available in this package
-	if b, err := Bytes(v); err == nil {
-		return b
-	}
-	// best-effort fallback
-	switch x := v.(type) {
-	case []byte:
-		return x
-	case string:
-		return []byte(x)
-	default:
-		return []byte(fmt.Sprintf("%v", v))
-	}
-}

messanger/messanger_mqtt.go

@@ -261,6 +261,7 @@ type MsgPrinter struct{}
 //	printer := &MsgPrinter{}
 //	msg.Subscribe("ss/c/+/temp", printer.MsgHandler)
 func (m *MsgPrinter) MsgHandler(msg *Msg) error {
-	fmt.Printf("%+v\n", msg)
+	str := fmt.Sprintf("%#v", msg)
+	slog.Info("builtin MsgHandler", "msg", str)
 	return nil
 }

messanger/messanger_test.go

@@ -48,11 +48,10 @@ func TestMessangerBaseServeHTTP(t *testing.T) {
 
 func TestNewMessanger(t *testing.T) {
 	tests := []struct {
-		name      string
-		id        string
-		topic     string
-		expected  string
-		expectNil bool
+		name     string
+		id       string
+		topic    string
+		expected string
 	}{
 		{
 			name:     "Create none messanger",
@@ -67,22 +66,18 @@ func TestNewMessanger(t *testing.T) {
 			expected: "mqtt",
 		},
 		{
-			name:      "Create messanger with unknown id returns nil",
-			id:        "unknown",
-			topic:     "topic1",
-			expectNil: true,
+			name:     "Create messanger with unknown id returns nil",
+			id:       "unknown",
+			topic:    "topic1",
+			expected: "unknown",
 		},
 	}
 
 	for _, tt := range tests {
 		t.Run(tt.name, func(t *testing.T) {
 			m := NewMessanger(tt.id)
-			if tt.expectNil {
-				assert.Nil(t, m)
-			} else {
-				assert.NotNil(t, m)
-				assert.Equal(t, tt.expected, m.ID())
-			}
+			assert.NotNil(t, m)
+			assert.Equal(t, tt.expected, m.ID())
 		})
 	}
 }

messanger/mqtt.go

@@ -41,14 +41,14 @@ var (
 //	}
 //	client.Publish("sensors/temp", "25.5")
 type MQTT struct {
-	id       string `json:"id"`       // Unique client identifier
-	Broker   string `json:"broker"`   // Broker hostname or IP (no protocol/port)
-	Username string `json:"username"` // MQTT authentication username
-	Password string `json:"password"` // MQTT authentication password
-	Debug    bool   `json:"debug"`    // Enable Paho MQTT client debug logging
+	id       string // Unique client identifier
+	Broker   string // Broker hostname or IP (no protocol/port)
+	Username string // MQTT authentication username
+	Password string // MQTT authentication password
+	Debug    bool   // Enable Paho MQTT client debug logging
 
-	error         `json:"error"` // Last error encountered
-	gomqtt.Client `json:"-"`     // Embedded Paho MQTT client
+	error         // Last error encountered
+	gomqtt.Client // Embedded Paho MQTT client
 }
 
 // NewMQTT creates a new MQTT client instance with default credentials.
@@ -69,8 +69,8 @@ func NewMQTT(id string, broker string, topics string) *MQTT {
 	mqtt = &MQTT{
 		id:       id,
 		Broker:   broker,
-		Username: "otto",
-		Password: "otto123",
+		Username: "",
+		Password: "",
 	}
 	return mqtt
 }

messanger/mqtt_broker.go

@@ -3,6 +3,8 @@ package messanger
 import (
 	"context"
 	"log"
+	"log/slog"
+	"sync"
 
 	mqttserver "github.com/mochi-mqtt/server/v2"
 	"github.com/mochi-mqtt/server/v2/hooks/auth"
@@ -52,6 +54,10 @@ import (
 // Optional Features (commented out):
 //   - WebSocket listener on port 1882 for browser clients
 //   - HTTP health check endpoint on port 8081 for monitoring
+var (
+	shutdown func(context.Context) error
+)
+
 func StartMQTTBroker(ctx context.Context) (func(context.Context) error, error) {
 	// Create broker with default options (in-memory state).
 	srv := mqttserver.New(nil)
@@ -107,15 +113,28 @@ func StartMQTTBroker(ctx context.Context) (func(context.Context) error, error) {
 		}
 	}()
 
+	var once sync.Once
+
+	// Return a shutdown function to be called from the app.
+	shutdown = func(_ context.Context) error {
+		slog.Debug("mqtt broker shutdown called")
+		// srv.Close() must be wrapped because it returns an error
+		once.Do(func() { srv.Close() })
+		return nil
+	}
+
 	// Close on context cancellation if provided.
 	go func() {
 		<-ctx.Done()
-		_ = srv.Close()
+		_ = shutdown(ctx)
 	}()
 
-	// Return a shutdown function you can call from your app.
-	shutdown := func(_ context.Context) error {
-		return srv.Close()
-	}
 	return shutdown, nil
 }
+
+func StopMQTTBroker(ctx context.Context) error {
+	if shutdown != nil {
+		shutdown(ctx)
+	}
+	return nil
+}