amqp

Documentation

Overview

Package amqp connects to an AMQP server in order to communicate with a gateway.

Connection/Disconnection of gateways is done by publishing messages to the "connect" and "disconnect" routing keys. When a gateway connects, it (or a plugin on the broker) should publish a protocol buffer of the type types.ConnectMessage containing the gateway's ID and either a key or a token to the "connect" routing key .

The gateway (or the plugin) can send a protocol buffer of the type types.DisconnectMessage containing the gateway's ID on the "disconnect" routing key when it disconnects, in order to help the bridge clean up connections.

Uplink messages are sent as protocol buffers on the "[gateway-id].up" routing key. The bridge should call `SubscribeUplink("gateway-id")` to subscribe to these. It is also possible to subscribe to a wildcard gateway by passing "*".

Downlink messages are sent as protocol buffers on the "[gateway-id].down" routing key. The bridge should call `PublishDownlink(*types.DownlinkMessage)` in order to send the downlink to the gateway.

Gateway status messages are sent as protocol buffers on the "[gateway-id].status" routing key. The bridge should call `SubscribeStatus("gateway-id")` to subscribe to these. It is also possible to subscribe to a wildcard gateway by passing "*".

Index

• Variables
• type AMQP
  • func New(config Config, ctx log.Interface) (*AMQP, error)
  • func (c *AMQP) Connect() error
  • func (c *AMQP) Disconnect() error
  • func (c *AMQP) Publish(routingKey string, message []byte) error
  • func (c *AMQP) PublishDownlink(message *types.DownlinkMessage) error
  • func (c *AMQP) SubscribeConnect() (<-chan *types.ConnectMessage, error)
  • func (c *AMQP) SubscribeDisconnect() (<-chan *types.DisconnectMessage, error)
  • func (c *AMQP) SubscribeStatus(gatewayID string) (<-chan *types.StatusMessage, error)
  • func (c *AMQP) SubscribeUplink(gatewayID string) (<-chan *types.UplinkMessage, error)
  • func (c *AMQP) UnsubscribeConnect() error
  • func (c *AMQP) UnsubscribeDisconnect() error
  • func (c *AMQP) UnsubscribeStatus(gatewayID string) error
  • func (c *AMQP) UnsubscribeUplink(gatewayID string) error
• type Config

Variables

var (
	ConnectRoutingKeyFormat    = "connect"
	DisconnectRoutingKeyFormat = "disconnect"
	UplinkRoutingKeyFormat     = "%s.up"
	DownlinkRoutingKeyFormat   = "%s.down"
	StatusRoutingKeyFormat     = "%s.status"
)

Routing Key formats for connect, disconnect, uplink, downlink and status messages

var (
	// ConnectRetries says how many times the client should retry a failed connection
	ConnectRetries = 10
	// ConnectRetryDelay says how long the client should wait between retries
	ConnectRetryDelay = time.Second
)

var BufferSize = 10

BufferSize indicates the maximum number of AMQP messages that should be buffered

Types

type AMQP

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

AMQP side of the bridge

func New

func New(config Config, ctx log.Interface) (*AMQP, error)

New returns a new AMQP

func (*AMQP) Connect

func (c *AMQP) Connect() error

Connect to AMQP

func (*AMQP) Disconnect

func (c *AMQP) Disconnect() error

Disconnect from AMQP

func (*AMQP) Publish

func (c *AMQP) Publish(routingKey string, message []byte) error

Publish a message to a routing key

func (*AMQP) PublishDownlink

func (c *AMQP) PublishDownlink(message *types.DownlinkMessage) error

PublishDownlink publishes a downlink message

func (*AMQP) SubscribeConnect

func (c *AMQP) SubscribeConnect() (<-chan *types.ConnectMessage, error)

SubscribeConnect subscribes to connect messages

func (*AMQP) SubscribeDisconnect

func (c *AMQP) SubscribeDisconnect() (<-chan *types.DisconnectMessage, error)

SubscribeDisconnect subscribes to disconnect messages

func (*AMQP) SubscribeStatus

func (c *AMQP) SubscribeStatus(gatewayID string) (<-chan *types.StatusMessage, error)

SubscribeStatus handles status messages for the given gateway ID

func (*AMQP) SubscribeUplink

func (c *AMQP) SubscribeUplink(gatewayID string) (<-chan *types.UplinkMessage, error)

SubscribeUplink handles uplink messages for the given gateway ID

func (*AMQP) UnsubscribeConnect

func (c *AMQP) UnsubscribeConnect() error

UnsubscribeConnect unsubscribes from connect messages

func (*AMQP) UnsubscribeDisconnect

func (c *AMQP) UnsubscribeDisconnect() error

UnsubscribeDisconnect unsubscribes from disconnect messages

func (*AMQP) UnsubscribeStatus

func (c *AMQP) UnsubscribeStatus(gatewayID string) error

UnsubscribeStatus unsubscribes from status messages for the given gateway ID

func (*AMQP) UnsubscribeUplink

func (c *AMQP) UnsubscribeUplink(gatewayID string) error

UnsubscribeUplink unsubscribes from uplink messages for the given gateway ID

type Config

type Config struct {
	Address        string
	Username       string
	Password       string
	VHost          string
	ExchangeName   string
	QueuePrefix    string
	ConsumerPrefix string
	TLSConfig      *tls.Config
}

Config contains configuration for AMQP