client

package module
v0.2.0 Latest Latest
Warning

This package is not in the latest version of its module.

 Go to latest Published: Jul 4, 2016 License: Apache-2.0 Imports: 13 Imported by: 0

README

gomqtt/client

Circle CI Coverage Status GoDoc Release Go Report Card

This go package implements functionality for communicating with a MQTT 3.1.1 broker.

Installation

Get it using go's standard toolset:

$ go get github.com/gomqtt/client

Usage

Service
wait := make(chan struct{})
done := make(chan struct{})

options := NewOptions()
options.ClientID = "gomqtt/service"

s := NewService()

s.Online = func(resumed bool) {
    fmt.Println("online!")
    fmt.Printf("resumed: %v\n", resumed)
}

s.Offline = func() {
    fmt.Println("offline!")
    close(done)
}

s.Message = func(msg *packet.Message) {
    fmt.Printf("message: %s - %s\n", msg.Topic, msg.Payload)
    close(wait)
}

ClearSession("mqtt://try:try@broker.shiftr.io", "gomqtt/service")

s.Start("mqtt://try:try@broker.shiftr.io", options)

s.Subscribe("test", 0, false).Wait()

s.Publish("test", []byte("test"), 0, false)

<-wait

s.Stop(true)

<-done

// Output:
// online!
// resumed: false
// message: test - test
// offline!
Client
done := make(chan struct{})

c := New()

c.Callback = func(msg *packet.Message, err error) {
    if err != nil {
        panic(err)
    }

    fmt.Printf("%s: %s\n", msg.Topic, msg.Payload)
    close(done)
}

options := NewOptions()
options.ClientID = "gomqtt/client"

connectFuture, err := c.Connect("mqtt://try:try@broker.shiftr.io", options)
if err != nil {
    panic(err)
}

err = connectFuture.Wait(10 * time.Second)
if err != nil {
    panic(err)
}

subscribeFuture, err := c.Subscribe("test", 0)
if err != nil {
    panic(err)
}

err = subscribeFuture.Wait(10 * time.Second)
if err != nil {
    panic(err)
}

publishFuture, err := c.Publish("test", []byte("test"), 0, false)
if err != nil {
    panic(err)
}

err = publishFuture.Wait(10 * time.Second)
if err != nil {
    panic(err)
}

<-done

err = c.Disconnect()
if err != nil {
    panic(err)
}

// Output:
// test: test

Documentation

Index

Examples

Constants

This section is empty.

Variables

View Source 
var ErrClientAlreadyConnecting = errors.New("client already connecting")

ErrClientAlreadyConnecting is returned by Connect if there has been already a connection attempt.

View Source 
var ErrClientConnectionDenied = errors.New("client connection denied")

ErrClientConnectionDenied is returned in the Callback if the connection has been reject by the broker.

View Source 
var ErrClientExpectedConnack = errors.New("client expected connack")

ErrClientExpectedConnack is returned when the first received packet is not a ConnackPacket.

View Source 
var ErrClientMissingID = errors.New("client missing id")

ErrClientMissingID is returned by Connect if no ClientID has been provided in the options while requesting to resume a session.

View Source 
var ErrClientMissingPong = errors.New("client missing pong")

ErrClientMissingPong is returned in the Callback if the broker did not respond in time to a PingreqPacket.

View Source 
var ErrClientNotConnected = errors.New("client not connected")

ErrClientNotConnected is returned by Publish, Subscribe and Unsubscribe if the client is not currently connected.

View Source 
var ErrClientUnexpectedClose = errors.New("client unexpected close")

ErrClientUnexpectedClose is returned in the Callback if the broker closed the connection without receiving a DisconnectPacket from the client.

View Source 
var ErrFutureCanceled = errors.New("future canceled")

ErrFutureCanceled is returned by Wait if the future gets canceled while waiting.

View Source 
var ErrFutureTimeout = errors.New("future timeout")

ErrFutureTimeout is returned by Wait if the specified timeout is exceeded.

Functions

func ClearRetainedMessage added in v0.2.0 

func ClearRetainedMessage(url string, topic string) error

ClearRetainedMessage will connect/disconnect and send an empty retained message. This is useful in situations where its not clear if a message has already been retained.

func ClearSession 

func ClearSession(url string, clientID string) error

ClearSession will connect/disconnect once with a clean session request to force the broker to reset the clients session. This is useful in situations where its not clear in what state the last session was left.

func SessionTest added in v0.2.0 

func SessionTest(t *testing.T, builder func() Session)

SessionTest will test a Session implementation. The passed builder callback should always return a fresh instances of the Session.

Types

type Callback 

type Callback func(msg *packet.Message, err error)

A Callback is a function called by the client upon received messages or internal errors.

type Client 

type Client struct {
	Session  Session
	Callback Callback
	Logger   Logger
	// contains filtered or unexported fields
}

A Client connects to a broker and handles the transmission of packets. It will automatically send PingreqPackets to keep the connection alive. Outgoing publish related packets will be stored in session and resend when the connection gets closed abruptly. All methods return Futures that get completed when the packets get acknowledged by the broker. Once the connection is closed all waiting futures get canceled.

Note: If clean session is false and there are packets in the store, messages might get completed after connecting without triggering any futures to complete.

Example 
done := make(chan struct{})

c := New()

c.Callback = func(msg *packet.Message, err error) {
	if err != nil {
		panic(err)
	}

	fmt.Printf("%s: %s\n", msg.Topic, msg.Payload)
	close(done)
}

options := NewOptions()
options.ClientID = "gomqtt/client"

connectFuture, err := c.Connect("mqtt://try:try@broker.shiftr.io", options)
if err != nil {
	panic(err)
}

err = connectFuture.Wait(10 * time.Second)
if err != nil {
	panic(err)
}

subscribeFuture, err := c.Subscribe("test", 0)
if err != nil {
	panic(err)
}

err = subscribeFuture.Wait(10 * time.Second)
if err != nil {
	panic(err)
}

publishFuture, err := c.Publish("test", []byte("test"), 0, false)
if err != nil {
	panic(err)
}

err = publishFuture.Wait(10 * time.Second)
if err != nil {
	panic(err)
}

<-done

err = c.Disconnect()
if err != nil {
	panic(err)
}

Output:
test: test

func New 

func New() *Client

New returns a new client that by default uses a fresh MemorySession.

func (*Client) Close 

func (c *Client) Close() error

Close closes the client immediately without sending a DisconnectPacket and waiting for outgoing transmissions to finish.

func (*Client) Connect 

func (c *Client) Connect(urlString string, opts *Options) (*ConnectFuture, error)

Connect opens the connection to the broker and sends a ConnectPacket. It will return a ConnectFuture that gets completed once a ConnackPacket has been received. If the ConnectPacket couldn't be transmitted it will return an error.

func (*Client) Disconnect 

func (c *Client) Disconnect(timeout ...time.Duration) error

Disconnect will send a DisconnectPacket and close the connection.

func (*Client) Publish 

func (c *Client) Publish(topic string, payload []byte, qos uint8, retain bool) (*PublishFuture, error)

Publish will send a PublishPacket containing the passed parameters. It will return a PublishFuture that gets completed once the quality of service flow has been completed.

func (*Client) PublishMessage 

func (c *Client) PublishMessage(msg *packet.Message) (*PublishFuture, error)

PublishMessage will send a PublishPacket containing the passed message. It will return a PublishFuture that gets completed once the quality of service flow has been completed.

func (*Client) Subscribe 

func (c *Client) Subscribe(topic string, qos uint8) (*SubscribeFuture, error)

Subscribe will send a SubscribePacket containing one topic to subscribe. It will return a SubscribeFuture that gets completed once a SubackPacket has been received.

func (*Client) SubscribeMultiple 

func (c *Client) SubscribeMultiple(subscriptions []packet.Subscription) (*SubscribeFuture, error)

SubscribeMultiple will send a SubscribePacket containing multiple topics to subscribe. It will return a SubscribeFuture that gets completed once a SubackPacket has been received.

func (*Client) Unsubscribe 

func (c *Client) Unsubscribe(topic string) (*UnsubscribeFuture, error)

Unsubscribe will send a UnsubscribePacket containing one topic to unsubscribe. It will return a UnsubscribeFuture that gets completed once a UnsubackPacket has been received.

func (*Client) UnsubscribeMultiple 

func (c *Client) UnsubscribeMultiple(topics []string) (*UnsubscribeFuture, error)

UnsubscribeMultiple will send a UnsubscribePacket containing multiple topics to unsubscribe. It will return a UnsubscribeFuture that gets completed once a UnsubackPacket has been received.

type ConnectFuture 

type ConnectFuture struct {
	SessionPresent bool
	ReturnCode     packet.ConnackCode
	// contains filtered or unexported fields
}

The ConnectFuture is returned by the Client on Connect.

func (*ConnectFuture) Call 

func (f *ConnectFuture) Call(callback func(error), timeout ...time.Duration)

func (*ConnectFuture) Wait 

func (f *ConnectFuture) Wait(timeout ...time.Duration) error

type Future 

type Future interface {
	// Wait will block until the future is completed or canceled. It will return
	// ErrCanceled  if the future gets canceled. If a timeout is specified it
	// might return a ErrTimeoutExceeded.
	Wait(timeout ...time.Duration) error

	// Call calls the supplied callback in a separate goroutine when Wait returns.
	Call(func(err error), ...time.Duration)
	// contains filtered or unexported methods
}

A Future represents information that might become available in the future.

type Logger 

type Logger func(msg string)

A Logger is a function called by the client to log activity.

type MemorySession added in v0.2.0 

type MemorySession struct {
	// contains filtered or unexported fields
}

A MemorySession stores packets in memory.

func NewMemorySession added in v0.2.0 

func NewMemorySession() *MemorySession

NewMemorySession returns a new MemorySession.

func (*MemorySession) AllPackets added in v0.2.0 

func (s *MemorySession) AllPackets(direction string) ([]packet.Packet, error)

AllPackets will return all packets currently saved in the session.

func (*MemorySession) DeletePacket added in v0.2.0 

func (s *MemorySession) DeletePacket(direction string, id uint16) error

DeletePacket will remove a packet from the session. The method must not return an error if no packet with the specified id does exists.

func (*MemorySession) LookupPacket added in v0.2.0 

func (s *MemorySession) LookupPacket(direction string, id uint16) (packet.Packet, error)

LookupPacket will retrieve a packet from the session using a packet id.

func (*MemorySession) PacketID added in v0.2.0 

func (s *MemorySession) PacketID() uint16

PacketID will return the next id for outgoing packets.

func (*MemorySession) Reset added in v0.2.0 

func (s *MemorySession) Reset() error

Reset will completely reset the session.

func (*MemorySession) SavePacket added in v0.2.0 

func (s *MemorySession) SavePacket(direction string, pkt packet.Packet) error

SavePacket will store a packet in the session. An eventual existing packet with the same id gets quietly overwritten.

type Message 

type Message func(msg *packet.Message)

Message is a function that is called when a message is received.

type Offline 

type Offline func()

Offline is a function that is called when the service is disconnected.

type Online 

type Online func(resumed bool)

Online is a function that is called when the service is connected.

type Options 

type Options struct {
	ClientID     string
	CleanSession bool
	KeepAlive    string
	Will         *packet.Message
}

Options are passed to a Client on Connect.

func NewOptions 

func NewOptions() *Options

NewOptions will initialize and return new Options.

type PublishFuture 

type PublishFuture struct {
	// contains filtered or unexported fields
}

The PublishFuture is returned by the Client on Publish.

func (*PublishFuture) Call 

func (f *PublishFuture) Call(callback func(error), timeout ...time.Duration)

func (*PublishFuture) Wait 

func (f *PublishFuture) Wait(timeout ...time.Duration) error

type Service 

type Service struct {
	Session Session
	Online  Online
	Message Message
	Offline Offline
	Logger  Logger

	MinReconnectDelay time.Duration
	MaxReconnectDelay time.Duration
	ConnectTimeout    time.Duration
	DisconnectTimeout time.Duration
	// contains filtered or unexported fields
}

Service is an abstraction for Client that provides a stable interface to the application, while it automatically connects and reconnects clients in the background. Errors are not returned but logged using the Logger callback. All methods return Futures that get completed once the acknowledgements are received. Once the services is stopped all waiting futures get canceled.

Note: If clean session is false and there are packets in the store, messages might get completed after starting without triggering any futures to complete.

Example 
wait := make(chan struct{})
done := make(chan struct{})

options := NewOptions()
options.ClientID = "gomqtt/service"

s := NewService()

s.Online = func(resumed bool) {
	fmt.Println("online!")
	fmt.Printf("resumed: %v\n", resumed)
}

s.Offline = func() {
	fmt.Println("offline!")
	close(done)
}

s.Message = func(msg *packet.Message) {
	fmt.Printf("message: %s - %s\n", msg.Topic, msg.Payload)
	close(wait)
}

ClearSession("mqtt://try:try@broker.shiftr.io", "gomqtt/service")

s.Start("mqtt://try:try@broker.shiftr.io", options)

s.Subscribe("test", 0, false).Wait()

s.Publish("test", []byte("test"), 0, false)

<-wait

s.Stop(true)

<-done

Output:
online!
resumed: false
message: test - test
offline!

func NewService 

func NewService() *Service

NewService allocates and returns a new service.

func (*Service) Publish 

func (s *Service) Publish(topic string, payload []byte, qos uint8, retain bool) *PublishFuture

Publish will send a PublishPacket containing the passed parameters. It will return a PublishFuture that gets completed once the quality of service flow has been completed.

func (*Service) Start 

func (s *Service) Start(url string, opts *Options)

Start will start the service with the specified configuration. From now on the service will automatically reconnect on any error until Stop is called.

func (*Service) Stop 

func (s *Service) Stop(clearFutures bool)

Stop will disconnect the client if online and cancel all futures if requested. After the service is stopped in can be started again.

Note: You should clear the futures on the last shutdown before exiting to ensure that all goroutines return that wait on futures.

func (*Service) Subscribe 

func (s *Service) Subscribe(topic string, qos uint8, requeue bool) *SubscribeFuture

Subscribe will send a SubscribePacket containing one topic to subscribe. If requeue is set to true the packet will be retried on an error.

func (*Service) SubscribeMultiple 

func (s *Service) SubscribeMultiple(subscriptions []packet.Subscription, requeue bool) *SubscribeFuture

SubscribeMultiple will send a SubscribePacket containing multiple topics to subscribe. If requeue is set to true the packet will be retried on an error.

func (*Service) Unsubscribe 

func (s *Service) Unsubscribe(topic string, requeue bool) *UnsubscribeFuture

Unsubscribe will send a UnsubscribePacket containing one topic to unsubscribe. If requeue is set to true the packet will be retried on an error.

func (*Service) UnsubscribeMultiple 

func (s *Service) UnsubscribeMultiple(topics []string, requeue bool) *UnsubscribeFuture

UnsubscribeMultiple will send a UnsubscribePacket containing multiple topics to unsubscribe. If requeue is set to true the packet will be retried on an error.

type Session added in v0.2.0 

type Session interface {
	// PacketID will return the next id for outgoing packets.
	PacketID() uint16

	// SavePacket will store a packet in the session. An eventual existing
	// packet with the same id gets quietly overwritten.
	SavePacket(direction string, pkt packet.Packet) error

	// LookupPacket will retrieve a packet from the session using a packet id.
	LookupPacket(direction string, id uint16) (packet.Packet, error)

	// DeletePacket will remove a packet from the session. The method must not
	// return an error if no packet with the specified id does exists.
	DeletePacket(direction string, id uint16) error

	// AllPackets will return all packets currently saved in the session.
	AllPackets(direction string) ([]packet.Packet, error)

	// Reset will completely reset the session.
	Reset() error
}

A Session is used to persist incoming and outgoing packets.

type SubscribeFuture 

type SubscribeFuture struct {
	ReturnCodes []uint8
	// contains filtered or unexported fields
}

The SubscribeFuture is returned by the Client on Subscribe.

func (*SubscribeFuture) Call 

func (f *SubscribeFuture) Call(callback func(error), timeout ...time.Duration)

func (*SubscribeFuture) Wait 

func (f *SubscribeFuture) Wait(timeout ...time.Duration) error

type UnsubscribeFuture 

type UnsubscribeFuture struct {
	// contains filtered or unexported fields
}

UnsubscribeFuture is returned by the Client on Unsubscribe.

func (*UnsubscribeFuture) Call 

func (f *UnsubscribeFuture) Call(callback func(error), timeout ...time.Duration)

func (*UnsubscribeFuture) Wait 

func (f *UnsubscribeFuture) Wait(timeout ...time.Duration) error

Source Files

View all Source files

Jump to

Keyboard shortcuts

? : This menu
/ : Search site
f or F : Jump to
y or Y : Canonical URL
go.dev uses cookies from Google to deliver and enhance the quality of its services and to analyze traffic. Learn more.