mqtt

func (*BigMessage) Error ¶

func (e *BigMessage) Error() string

Error implements the standard error interface.

func (*BigMessage) ReadAll ¶

func (e *BigMessage) ReadAll() ([]byte, error)

ReadAll returns the message in a new/dedicated buffer. Messages can be read only once, after reception (from ReadSlices), and before the next ReadSlices. The invocation must occur from within the same routine.

func (*Client) Close ¶

func (c *Client) Close() error

Close terminates the connection establishment. The Client is closed regardless of the error return. Closing an already closed Client has no effect.

func (*Client) Disconnect ¶

func (c *Client) Disconnect(quit <-chan struct{}) error

Disconnect tries a graceful termination, which discards the Will. The Client is closed regardless of the error return.

Quit is optional, as nil just blocks. Appliance of quit will strictly result in ErrCanceled.

BUG(pascaldekloe): The MQTT protocol has no confirmation for the disconnect request. As a result, a client can never know for sure whether the operation actually succeeded.

func (*Client) Offline ¶

func (c *Client) Offline() <-chan struct{}

Offline returns a chanel that's closed when the client has no connection.

func (*Client) Online ¶

func (c *Client) Online() <-chan struct{}

Online returns a chanel that's closed when the client has a connection.

func (*Client) Ping ¶

func (c *Client) Ping(quit <-chan struct{}) error

Ping makes a roundtrip to validate the connection. Only one request is permitted ErrMax at a time.

Quit is optional, as nil just blocks. Appliance of quit will strictly result in either ErrCanceled or ErrAbandoned.

func (*Client) Publish ¶

func (c *Client) Publish(quit <-chan struct{}, message []byte, topic string) error

Publish delivers the message with an “at most once” guarantee. Subscribers may or may not receive the message when subject to error. This delivery method is the most efficient option.

Quit is optional, as nil just blocks. Appliance of quit will strictly result in ErrCanceled.

func (*Client) PublishAtLeastOnce ¶

func (c *Client) PublishAtLeastOnce(message []byte, topic string) (exchange <-chan error, err error)

PublishAtLeastOnce delivers the message with an “at least once” guarantee. Subscribers may receive the message more than once when subject to error. This delivery method requires a response transmission plus persistence on both client-side and broker-side.

The exchange channel is closed uppon receival confirmation by the broker. ErrClosed leaves the channel blocked (with no further input).

package main

import (
	"errors"
	"fmt"
	"time"

	"github.com/pascaldekloe/mqtt"
)

// PublishAtLeastOnce is a method from mqtt.Client.
var PublishAtLeastOnce func(message []byte, topic string) (ack <-chan error, err error)

func main() {
	for {
		exchange, err := PublishAtLeastOnce([]byte("🍸🆘"), "demo/alert")
		switch {
		case err == nil:
			fmt.Println("alert submitted…")
			break

		case mqtt.IsDeny(err), errors.Is(err, mqtt.ErrClosed):
			fmt.Println("🚨 alert not send:", err)
			return

		case errors.Is(err, mqtt.ErrMax):
			fmt.Println("⚠️ alert submission hold-up:", err)
			time.Sleep(time.Second / 4)
			continue

		default:
			fmt.Println("⚠️ alert submission blocked on persistence malfunction:", err)
			time.Sleep(4 * time.Second)
			continue
		}

		for err := range exchange {
			if errors.Is(err, mqtt.ErrClosed) {
				fmt.Println("🚨 alert exchange suspended:", err)
				// An AdoptSession may continue the transaction.
				return
			}

			fmt.Println("⚠️ alert request transfer interrupted:", err)
		}
		fmt.Println("alert acknowledged ✓")
		break
	}

}

Output:

alert submitted…
alert acknowledged ✓

func (*Client) PublishAtLeastOnceRetained ¶

func (c *Client) PublishAtLeastOnceRetained(message []byte, topic string) (exchange <-chan error, err error)

PublishAtLeastOnceRetained is like PublishAtLeastOnce, but the broker must store the message, so that it can be delivered to future subscribers whose subscriptions match the topic name. When a new subscription is established, the last retained message, if any, on each matching topic name must be sent to the subscriber.

func (*Client) PublishExactlyOnce ¶

func (c *Client) PublishExactlyOnce(message []byte, topic string) (exchange <-chan error, err error)

PublishExactlyOnce delivers the message with an “exactly once” guarantee. This delivery method eliminates the duplicate-delivery risk from PublishAtLeastOnce at the expense of an additional network roundtrip.

func (*Client) PublishExactlyOnceRetained ¶

func (c *Client) PublishExactlyOnceRetained(message []byte, topic string) (exchange <-chan error, err error)

PublishExactlyOnceRetained is like PublishExactlyOnce, but the broker must store the message, so that it can be delivered to future subscribers whose subscriptions match the topic name. When a new subscription is established, the last retained message, if any, on each matching topic name must be sent to the subscriber.

func (*Client) PublishRetained ¶

func (c *Client) PublishRetained(quit <-chan struct{}, message []byte, topic string) error

PublishRetained is like Publish, but the broker must store the message, so that it can be delivered to future subscribers whose subscriptions match the topic name. The broker may choose to discard the message at any time though. Uppon reception, the broker must discard any message previously retained for the topic name.

func (*Client) ReadSlices ¶

func (c *Client) ReadSlices() (message, topic []byte, err error)

ReadSlices should be invoked consecutively from a single goroutine until ErrClosed. Both message and topic are slices from a read buffer. The bytes stop being valid at the next read. Care must be taken to avoid freezing the Client. Use any of the following techniques for blocking operations.

• start a goroutine with a copy of message and/or topic
• start a goroutine with the bytes parsed/unmarshalled
• persist message and/or topic; then continue from there
• apply low timeouts in a strict manner

Each invocation acknowledges ownership of the previous returned, if any, including BigMessage.

BigMessage leaves memory allocation beyond the read buffer as a choice to the consumer. Any error other than BigMessage puts the Client in an ErrDown state. Invocation should apply a backoff once down. Retries on IsConnectionRefused, if any, should probably apply a rather large backoff. See the Client example for a complete setup.

func (*Client) Subscribe ¶

func (c *Client) Subscribe(quit <-chan struct{}, topicFilters ...string) error

Subscribe requests subscription for all topics that match any of the filter arguments.

Quit is optional, as nil just blocks. Appliance of quit will strictly result in either ErrCanceled or ErrAbandoned.

package main

import (
	"context"
	"errors"
	"fmt"
	"time"

	"github.com/pascaldekloe/mqtt"
)

// Subscribe is a method from mqtt.Client.
var Subscribe func(quit <-chan struct{}, topicFilters ...string) error

// Online is a method from mqtt.Client.
var Online func() <-chan struct{}

func main() {
	ctx, cancel := context.WithTimeout(context.Background(), time.Minute)
	defer cancel()

	for {
		err := Subscribe(ctx.Done(), "demo/+")
		switch {
		case err == nil:
			fmt.Println("subscribe confirmed by broker")
			return

		case errors.As(err, new(mqtt.SubscribeError)):
			fmt.Println("subscribe failed by broker")
			return

		case mqtt.IsDeny(err): // illegal topic filter
			fmt.Println(err)
			return

		case errors.Is(err, mqtt.ErrClosed):
			fmt.Println("no subscribe due client close")
			return

		case errors.Is(err, mqtt.ErrCanceled):
			fmt.Println("no subscribe due timeout")
			return

		case errors.Is(err, mqtt.ErrAbandoned):
			fmt.Println("subscribe state unknown due timeout")
			return

		case errors.Is(err, mqtt.ErrBreak):
			fmt.Println("subscribe state unknown due connection loss")
			select {
			case <-Online():
				fmt.Println("subscribe retry with new connection")
			case <-ctx.Done():
				fmt.Println("subscribe timeout")
				return
			}

		case errors.Is(err, mqtt.ErrDown):
			fmt.Println("subscribe delay while service is down")
			select {
			case <-Online():
				fmt.Println("subscribe retry with new connection")
			case <-ctx.Done():
				fmt.Println("subscribe timeout")
				return
			}

		case errors.Is(err, mqtt.ErrMax):
			fmt.Println("subscribe hold-up due excessive number of pending requests")
			time.Sleep(2 * time.Second) // backoff

		default:
			fmt.Println("subscribe request transfer interrupted:", err)
			time.Sleep(time.Second / 2) // backoff
		}
	}
}

Output:

subscribe confirmed by broker

func (*Client) SubscribeLimitAtLeastOnce ¶

func (c *Client) SubscribeLimitAtLeastOnce(quit <-chan struct{}, topicFilters ...string) error

SubscribeLimitAtLeastOnce is like Subscribe, but it limits message reception to quality-of-service level 1—acknowledged transfer.

func (*Client) SubscribeLimitAtMostOnce ¶

func (c *Client) SubscribeLimitAtMostOnce(quit <-chan struct{}, topicFilters ...string) error

SubscribeLimitAtMostOnce is like Subscribe, but it limits message reception to quality-of-service level 0—fire and forget.

func (*Client) Unsubscribe ¶

func (c *Client) Unsubscribe(quit <-chan struct{}, topicFilters ...string) error

Unsubscribe requests subscription cancelation for each of the filter arguments.

Quit is optional, as nil just blocks. Appliance of quit will strictly result in either ErrCanceled or ErrAbandoned.

func (SubscribeError) Error ¶

func (e SubscribeError) Error() string

Error implements the standard error interface.