mqtt-5-learning/internal/mqtt/client.go

package mqtt

import (
	"context"
	"fmt"
	"log"
	"net/url"
	"time"

	"example.com/mqtt5-learning/internal/config"
	"example.com/mqtt5-learning/internal/contracts"
	"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 []contracts.TopicFilter, 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.String(), 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 contracts.TopicName, payload []byte) error {
	operation, cancel := context.WithTimeout(ctx, 3*time.Second)
	defer cancel()

	ack, err := manager.Publish(operation, &paho.Publish{
		Topic:   topic.String(),
		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)
	}
}