updated and tested till end of lesson 4

internal/mqtt/client.go

@@ -0,0 +1,143 @@
+package mqtt
+
+import (
+	"context"
+	"fmt"
+	"log"
+	"net/url"
+	"time"
+
+	"example.com/mqtt5-learning/internal/config"
+	"github.com/eclipse/paho.golang/autopaho"
+	"github.com/eclipse/paho.golang/paho"
+)
+
+type Message struct {
+	Topic   string
+	Payload []byte
+	Retain  bool
+}
+
+func Start(ctx context.Context, cfg config.Config, topics []string, inbox chan<- Message) (*autopaho.ConnectionManager, error) {
+	broker, err := url.Parse(cfg.MQTT.URL)
+	if err != nil {
+		return nil, err
+	}
+	initialSubscription := make(chan error, 1)
+	manager, err := autopaho.NewConnection(ctx, autopaho.ClientConfig{
+		ServerUrls:                    []*url.URL{broker},
+		KeepAlive:                     20,
+		CleanStartOnInitialConnection: true,
+		SessionExpiryInterval:         0,
+		ConnectUsername:               cfg.MQTT.Username,
+		ConnectPassword:               []byte(cfg.MQTT.Password),
+		OnConnectionUp: func(cm *autopaho.ConnectionManager, _ *paho.Connack) {
+			// This runs again on reconnect; reestablish broker subscriptions.
+			subCtx, cancel := context.WithTimeout(ctx, 5*time.Second)
+			defer cancel()
+
+			subscriptions := make([]paho.SubscribeOptions, 0, len(topics))
+			for _, topic := range topics {
+				subscriptions = append(subscriptions, paho.SubscribeOptions{Topic: topic, QoS: 1, RetainAsPublished: true})
+			}
+
+			ack, subErr := cm.Subscribe(subCtx, &paho.Subscribe{Subscriptions: subscriptions})
+			if subErr == nil {
+				if ack == nil || len(ack.Reasons) != len(topics) {
+					subErr = fmt.Errorf("missing or incomplete SUBACK")
+				} else {
+					for i, reason := range ack.Reasons {
+						if reason > 1 { // We requested QoS 1; valid grants are 0 or 1.
+							subErr = fmt.Errorf("subscription %s rejected: 0x%02x", topics[i], reason)
+							break
+						}
+					}
+				}
+			}
+			if subErr != nil {
+				log.Printf("subscribe failed: %v", subErr)
+			} else {
+				log.Printf("subscriptions ready: %v", topics)
+			}
+			select {
+			case initialSubscription <- subErr:
+			default:
+			}
+		},
+		OnConnectionDown: func() bool { log.Print("disconnected; reconnecting"); return true },
+		OnConnectError:   func(err error) { log.Printf("connect error: %v", err) },
+			ClientID:           cfg.MQTT.ClientID,
+			OnClientError:      func(err error) { log.Printf("client error: %v", err) },
+			OnServerDisconnect: func(d *paho.Disconnect) { log.Printf("disconnect reason=0x%02x", d.ReasonCode) },
+			OnPublishReceived: []func(paho.PublishReceived) (bool, error){
+				func(received paho.PublishReceived) (bool, error) {
+					packet := received.Packet
+					message := Message{Topic: packet.Topic, Payload: append([]byte(nil), packet.Payload...), Retain: packet.Retain}
+					select {
+					case inbox <- message:
+					default:
+						// Deliberately lossy early demo: automatic MQTT ACK is still possible.
+						log.Printf("lab inbox full; message dropped topic=%s", packet.Topic)
+					}
+					return true, nil
+				},
+			},
+	})
+	if err != nil {
+		return nil, fmt.Errorf("start AutoPaho: %w", err)
+	}
+	startup, cancel := context.WithTimeout(ctx, cfg.ConnectTimeout())
+	defer cancel()
+	if err = manager.AwaitConnection(startup); err != nil {
+		Stop(manager)
+		return nil, fmt.Errorf("await connection: %w", err)
+	}
+	select {
+	case err = <-initialSubscription:
+		if err != nil {
+			Stop(manager)
+			return nil, err
+		}
+	case <-startup.Done():
+		Stop(manager)
+		return nil, fmt.Errorf("await subscriptions: %w", startup.Err())
+	}
+	return manager, nil
+}
+
+func Publish(ctx context.Context, manager *autopaho.ConnectionManager, topic string, payload []byte) error {
+	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
+	defer cancel()
+
+	ack, err := manager.Publish(operation, &paho.Publish{
+		Topic:   topic,
+		Payload: payload,
+		QoS:     1,
+		Retain:  false,
+		Properties: &paho.PublishProperties{
+			ContentType: "application/json",
+		},
+	})
+	if err != nil {
+		return fmt.Errorf("publish: %w", err)
+	}
+	if ack == nil {
+		return fmt.Errorf("missing Qos 1 publication response")
+	}
+	if ack.ReasonCode >= 0x80 {
+		return fmt.Errorf("publish rejected: 0x%02x", ack.ReasonCode)
+	}
+
+	log.Printf("broker publish acknowledgement topic=%s reason=0x%02x", topic, ack.ReasonCode)
+
+	return nil
+}
+
+func Stop(manager *autopaho.ConnectionManager) {
+	shutdown, cancel := context.WithTimeout(context.Background(), 5*time.Second)
+	defer cancel()
+
+	if err:=manager.Disconnect(shutdown); err != nil {
+		log.Printf("shutdown did not finish within deadline: %v", err)
+	}
+}