ditto

README

Eclipse Ditto Client SDK for Golang

This repository contains the Golang client SDK for Eclipse Ditto.

Currently, Eclipse Hono MQTT is the only one supported transport.

Table of Contents

• Installation
• Creating and connecting a client
• Working with features
  • Create a new feature instance
  • Modify a feature's property
• Subscribing and handling messages

Installation

go get github.com/eclipse/ditto-clients-golang

Creating and connecting a client

Each client instance requires a ditto.Configuration object.

config := ditto.NewConfiguration().
    WithKeepAlive(30 * time.Second). // default keep alive is 30 seconds
    // WithCredentials(&ditto.Credentials{Username: "John", Password: "qwerty"}). if such are available or required
    WithBroker("mqtt-host:1883").
    WithConnectHandler(connectHandler)

func connectHandler(client ditto.Client) {
    // add logic to be executed when the client is connected
}

With this configuration a client instance could be created.

client = ditto.NewClient(config)

NOTE: In some cases an external Paho instance could be provided for the communication. If this is the case, there is a ditto.NewClientMQTT() create function available.

After you have configured and created your client instance, it's ready to be connected.

if err := client.Connect(); err != nil {
    panic(fmt.Errorf("cannot connect to broker: %v", err))
}
defer disconnect(client)

Working with features

Create a new feature instance

Define the feature to be created.

myFeature := &model.Feature{}
myFeature.
    WithDefinitionFrom("my.model.namespace:FeatureModel:1.0.0"). // you can provide a semantic definition of your feature
    WithProperty("myProperty", "myValue")

Create your Ditto command. Modify acts as an upsert - it either updates or creates features.

command := things.
    NewCommand(model.NewNamespacedIDFrom("my.namespace:thing.id")). // specify which thing you will send the command to
    Twin().
    Feature("MyFeature").
    Modify(myFeature) // the payload for the modification - i.e. the feature's JSON representation

Send the Ditto command.

envelope := command.Envelope(protocol.WithResponseRequired(false))
if err := client.Send(envelope); err != nil {
    fmt.Printf("could not send Ditto message: %v\n", err)
}

Modify a feature's property

Modify overrides the current feature's property.

command = things.
    NewCommand(model.NewNamespacedIDFrom("my.namespace:thing.id")). // specify which thing you will send the command to
    Twin().
    FeatureProperty("MyFeature", "myProperty").
    Modify("myNewValue") // the payload for the modification - i.e. the new property's value JSON representation

Subscribing and handling messages

Subscribe for incoming Ditto messages.

func connectHandler(client ditto.Client) {
    // it's a good practise to subscribe after the client is connected
    client.Subscribe(messagesHandler)
}

NOTE: You can add multiple handlers for Ditto messages processing.

It's a good practice to clear all subscriptions on client disconnect.

func disconnect(client ditto.Client) {
    // add any resources clearing logic
    client.Unsubscribe()
    client.Disconnect()
}

NOTE: If no message handler is provided then all would be removed.

Handle and reply to Ditto messages.

func messagesHandler(requestID string, msg *protocol.Envelope) {
    if msg.Topic.Namespace == "my.namespace" && msg.Topic.EntityID == "thing.id" &&
            msg.Path == "/features/MyFeature/inbox/messages/myCommand" {
        // respond to the message by using the outbox
        response := things.NewMessage(model.NewNamespacedID(msg.Topic.Namespace, msg.Topic.EntityID)).
            Feature("MyFeature").Outbox("myCommand").WithPayload("responsePayload")
        responseMsg := response.Envelope(protocol.WithCorrelationID(msg.Headers.CorrelationID()), protocol.WithResponseRequired(false))
        responseMsg.Status = 200
        if replyErr := client.Reply(requestID, responseMsg); replyErr != nil {
            fmt.Printf("failed to send response to request Id %s: %v\n", requestID, replyErr)
        }
    }
}

Logging

A custom logger could be implemented based on ditto.Logger interface. For example:

type logger struct {
	prefix string
}

func (l logger) Println(v ...interface{}) {
    fmt.Println(l.prefix, fmt.Sprint(v...))
}

func (l logger) Printf(format string, v ...interface{}) {
    fmt.Printf(fmt.Sprint(l.prefix, " ", format), v...)
}

Then the Ditto library could be configured to use the logger by assigning the logging endpoints - ERROR, WARN, INFO and DEBUG.

func init() {
    ditto.ERROR = logger{prefix: "ERROR  "}
    ditto.WARN  = logger{prefix: "WARN   "}
    ditto.INFO  = logger{prefix: "INFO   "}
    ditto.DEBUG = logger{prefix: "DEBUG  "}
}

Documentation

Index

• Variables
• type Client
  • func NewClient(cfg *Configuration) Client
  • func NewClientMQTT(mqttClient MQTT.Client, cfg *Configuration) (Client, error)
• type Configuration
  • func NewConfiguration() *Configuration
  • func (cfg *Configuration) AcknowledgeTimeout() time.Duration
  • func (cfg *Configuration) Broker() string
  • func (cfg *Configuration) ConnectHandler() ConnectHandler
  • func (cfg *Configuration) ConnectTimeout() time.Duration
  • func (cfg *Configuration) ConnectionLostHandler() ConnectionLostHandler
  • func (cfg *Configuration) Credentials() *Credentials
  • func (cfg *Configuration) DisconnectTimeout() time.Duration
  • func (cfg *Configuration) KeepAlive() time.Duration
  • func (cfg *Configuration) SubscribeTimeout() time.Duration
  • func (cfg *Configuration) TLSConfig() *tls.Config
  • func (cfg *Configuration) UnsubscribeTimeout() time.Duration
  • func (cfg *Configuration) WithAcknowledgeTimeout(acknowledgeTimeout time.Duration) *Configuration
  • func (cfg *Configuration) WithBroker(broker string) *Configuration
  • func (cfg *Configuration) WithConnectHandler(connectHandler ConnectHandler) *Configuration
  • func (cfg *Configuration) WithConnectTimeout(connectTimeout time.Duration) *Configuration
  • func (cfg *Configuration) WithConnectionLostHandler(connectionLostHandler ConnectionLostHandler) *Configuration
  • func (cfg *Configuration) WithCredentials(credentials *Credentials) *Configuration
  • func (cfg *Configuration) WithDisconnectTimeout(disconnectTimeout time.Duration) *Configuration
  • func (cfg *Configuration) WithKeepAlive(keepAlive time.Duration) *Configuration
  • func (cfg *Configuration) WithSubscribeTimeout(subscribeTimeout time.Duration) *Configuration
  • func (cfg *Configuration) WithTLSConfig(tlsConfig *tls.Config) *Configuration
  • func (cfg *Configuration) WithUnsubscribeTimeout(unsubscribeTimeout time.Duration) *Configuration
• type ConnectHandler
• type ConnectionLostHandler
• type Credentials
• type Handler
• type Logger
• type LoggerStub
  • func (LoggerStub) Printf(format string, v ...interface{})
  • func (LoggerStub) Println(v ...interface{})

Variables

var (
	// ErrAcknowledgeTimeout is an error that acknowledgement is not received within the timeout.
	ErrAcknowledgeTimeout = errors.New("acknowledge timeout")
	// ErrSubscribeTimeout is an error that subscription confirmation is not received within the timeout.
	ErrSubscribeTimeout = errors.New("subscribe timeout")
	// ErrUnsubscribeTimeout is an error that unsubscription confirmation is not received within the timeout.
	ErrUnsubscribeTimeout = errors.New("unsubscribe timeout")
)

Types

type Client

type Client interface {

	// Connect connects the client to the configured Ditto endpoint provided via the Client's Configuration at creation time.
	// If any error occurs during the connection's initiation - it's returned here.
	// An actual connection status is callbacked to the provided ConnectHandler
	// as soon as the connection is established and all Client's internal preparations are performed.
	// If the connection gets lost during runtime - the ConnectionLostHandler is notified to handle the case.
	Connect() error

	// Disconnect disconnects the client from the configured Ditto endpoint.
	Disconnect()

	// Reply is an auxiliary method to send replies for specific requestIDs if such has been provided along with the incoming protocol.Envelope.
	// The requestID must be the same as the one provided with the request protocol.Envelope.
	// An error is returned if the reply could not be sent for some reason.
	Reply(requestID string, message *protocol.Envelope) error

	// Send sends a protocol.Envelope to the Client's configured Ditto endpoint.
	// An error is returned if the envelope could not be sent for some reason.
	Send(message *protocol.Envelope) error

	// Subscribe ensures that all incoming Ditto messages will be transferred to the provided Handlers.
	Subscribe(handlers ...Handler)

	// Unsubscribe cancels sending incoming Ditto messages from the client to the provided Handlers
	// and removes them from the subscriptions list of the client.
	// If Unsubscribe is called without arguments, it will cancel and remove all currently subscribed Handlers.
	Unsubscribe(handlers ...Handler)
}

Client is the Ditto's library main interface definition. The interface is intended to abstract multiple implementations over different transports. Client has connect/disconnect capabilities along with the options to subscribe/unsubscribe for receiving all Ditto messages being exchanged using the underlying transport.

func NewClient

func NewClient(cfg *Configuration) Client

NewClient creates a new Client instance with the provided Configuration.

func NewClientMQTT

func NewClientMQTT(mqttClient MQTT.Client, cfg *Configuration) (Client, error)

NewClientMQTT creates a new Client instance with the Configuration, if such is provided, that is going to use the external MQTT client.

It is expected that the provided MQTT client is already connected. So this Client must be controlled from outside and its Connect/Disconnect methods must be invoked accordingly.

If a Configuration is provided it may include ConnectHandler and ConnectionLostHandler, as well as acknowledge, subscribe and unsubscribe timeout. As an external MQTT client is used, other fields are not needed and regarded as invalid ones.

Returns an error if the provided MQTT client is not connected or the Configuration contains invalid fields.

type Configuration

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

Configuration provides the Client's configuration.

func NewConfiguration

func NewConfiguration() *Configuration

NewConfiguration creates a new Configuration instance.

func (*Configuration) AcknowledgeTimeout

func (cfg *Configuration) AcknowledgeTimeout() time.Duration

AcknowledgeTimeout provides the timeout to wait for confirmation that a MQTT message is delivered to the broker. The default is 15 seconds.

func (*Configuration) Broker

func (cfg *Configuration) Broker() string

Broker provides the current MQTT broker the client is to connect to.

func (*Configuration) ConnectHandler

func (cfg *Configuration) ConnectHandler() ConnectHandler

ConnectHandler provides the currently configured ConnectHandler.

func (*Configuration) ConnectTimeout

func (cfg *Configuration) ConnectTimeout() time.Duration

ConnectTimeout provides the timeout for connecting the client. The default is 30 seconds.

func (*Configuration) ConnectionLostHandler

func (cfg *Configuration) ConnectionLostHandler() ConnectionLostHandler

ConnectionLostHandler provides the currently configured ConnectionLostHandler.

func (*Configuration) Credentials

func (cfg *Configuration) Credentials() *Credentials

Credentials provides the currently configured authentication credentials used for the underlying connection.

func (*Configuration) DisconnectTimeout

func (cfg *Configuration) DisconnectTimeout() time.Duration

DisconnectTimeout provides the timeout for disconnecting the client. The default is 250 milliseconds.

func (*Configuration) KeepAlive

func (cfg *Configuration) KeepAlive() time.Duration

KeepAlive provides the keep alive connection's period. The default is 30 seconds.

func (*Configuration) SubscribeTimeout

func (cfg *Configuration) SubscribeTimeout() time.Duration

SubscribeTimeout provides the timeout to wait for successful MQTT subscription. The default is 15 seconds.

func (*Configuration) TLSConfig

func (cfg *Configuration) TLSConfig() *tls.Config

TLSConfig provides the current TLS configuration for the underlying connection.

func (*Configuration) UnsubscribeTimeout

func (cfg *Configuration) UnsubscribeTimeout() time.Duration

UnsubscribeTimeout provides the timeout to wait for successful MQTT unsubscription. The default is 5 seconds.

func (*Configuration) WithAcknowledgeTimeout

func (cfg *Configuration) WithAcknowledgeTimeout(acknowledgeTimeout time.Duration) *Configuration

WithAcknowledgeTimeout configures the timeout for acknowledgement of the Client.

func (*Configuration) WithBroker

func (cfg *Configuration) WithBroker(broker string) *Configuration

WithBroker configures the MQTT's broker the Client to connect to.

func (*Configuration) WithConnectHandler

func (cfg *Configuration) WithConnectHandler(connectHandler ConnectHandler) *Configuration

WithConnectHandler configures the connectHandler to be notified when the Client's connection is established.

func (*Configuration) WithConnectTimeout

func (cfg *Configuration) WithConnectTimeout(connectTimeout time.Duration) *Configuration

WithConnectTimeout configures the timeout for connection of the Client.

func (*Configuration) WithConnectionLostHandler

func (cfg *Configuration) WithConnectionLostHandler(connectionLostHandler ConnectionLostHandler) *Configuration

WithConnectionLostHandler configures the connectionLostHandler to be notified is the Client's connection gets lost during runtime.

func (*Configuration) WithCredentials

func (cfg *Configuration) WithCredentials(credentials *Credentials) *Configuration

WithCredentials configures the credentials to be used for authentication by the underlying connection of the Client.

func (*Configuration) WithDisconnectTimeout

func (cfg *Configuration) WithDisconnectTimeout(disconnectTimeout time.Duration) *Configuration

WithDisconnectTimeout configures the timeout for disconnection of the Client.

func (*Configuration) WithKeepAlive

func (cfg *Configuration) WithKeepAlive(keepAlive time.Duration) *Configuration

WithKeepAlive configures the keep alive time period for the underlying Client's connection.

func (*Configuration) WithSubscribeTimeout

func (cfg *Configuration) WithSubscribeTimeout(subscribeTimeout time.Duration) *Configuration

WithSubscribeTimeout configures the timeout for subscription of the Client.

func (*Configuration) WithTLSConfig

func (cfg *Configuration) WithTLSConfig(tlsConfig *tls.Config) *Configuration

WithTLSConfig sets the TLS configuration to be used by the Client's underlying connection.

func (*Configuration) WithUnsubscribeTimeout

func (cfg *Configuration) WithUnsubscribeTimeout(unsubscribeTimeout time.Duration) *Configuration

WithUnsubscribeTimeout configures the timeout for unsubscription of the Client.

type ConnectHandler

type ConnectHandler func(client Client)

ConnectHandler is called when a successful connection to the configured Ditto endpoint is established and all Client's internal preparations are done.

type ConnectionLostHandler

type ConnectionLostHandler func(client Client, err error)

ConnectionLostHandler is called is the connection is lost during runtime.

type Credentials

type Credentials struct {
	Username string
	Password string
}

Credentials represents a user credentials for authentication used by the underlying connection (e.g. MQTT).

type Handler

type Handler func(requestID string, message *protocol.Envelope)

Handler represents a callback handler that is called on each received message. If the underlying transport provides a special requestID related to the Envelope, it's also provided to the handler so that chained responses to the ID can be later sent properly.

type Logger

type Logger interface {
	Println(v ...interface{})
	Printf(format string, v ...interface{})
}

Logger interface allows plugging of a logger implementation that fits best the needs of the application that is to use the Ditto library.

var (
	INFO  Logger = LoggerStub{}
	WARN  Logger = LoggerStub{}
	DEBUG Logger = LoggerStub{}
	ERROR Logger = LoggerStub{}
)

Levels of the library's output that can be configured during package initialization in init().

type LoggerStub

type LoggerStub struct{}

LoggerStub provides an empty default implementation.

func (LoggerStub) Printf

func (LoggerStub) Printf(format string, v ...interface{})

Printf provides an empty default implementation for formatted logging.

func (LoggerStub) Println

func (LoggerStub) Println(v ...interface{})

Println provides an empty default implementation for logging.