broker

README

Broker

This package abstracts the embedded message queue broker.

Features:

• nats server
• mochi-co mqtt

Examples:

package main

import (
	"fmt"
	"github.com/zourva/pareto/broker"
	"time"
)

func main() {
	// TODO
}

Documentation

Index

• type EmbeddedNats
  • func NewEmbeddedNats(opts ...Option) (*EmbeddedNats, error)
  • func (s *EmbeddedNats) Shutdown() error
  • func (s *EmbeddedNats) Startup() error
• type MQ
• type MQClient
• type MQClientOptions
• type MQTTClient
  • func NewMQTTClient(network string, endpoint string) *MQTTClient
  • func (c *MQTTClient) Connect(connMsg *paho.Connect) error
  • func (c *MQTTClient) Disconnect() error
• type MQTTClientV3
  • func NewMQTTClientV3(network string, endpoint string) *MQTTClientV3
  • func (c *MQTTClientV3) Connect(connMsg *paho.Connect) error
  • func (c *MQTTClientV3) Disconnect() error
• type MQTTServer
  • func NewMQTTServer(name string, endpoint string) *MQTTServer
  • func (s *MQTTServer) Shutdown() error
  • func (s *MQTTServer) Startup() error
• type Option
  • func WithAuthorizationToken(t string) Option
  • func WithHost(host string) Option
  • func WithLoggerFile(file string) Option
  • func WithMonitorHost(host string) Option
  • func WithMonitorPort(port int) Option
  • func WithPort(port int) Option
  • func WithRetryCount(c int) Option
  • func WithRetryDuration(d time.Duration) Option

Types

type EmbeddedNats

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

EmbeddedNats embeds the nats MQ server.

see https://docs.nats.io/running-a-nats-service/clients#embedding-nats and https://dev.to/karanpratapsingh/embedding-nats-in-go-19o.

func NewEmbeddedNats

func NewEmbeddedNats(opts ...Option) (*EmbeddedNats, error)

NewEmbeddedNats creates and initialize an embedded nats server using the given options.

func (*EmbeddedNats) Shutdown

func (s *EmbeddedNats) Shutdown() error

Shutdown stops the broker.

func (*EmbeddedNats) Startup

func (s *EmbeddedNats) Startup() error

Startup starts the broker.

type MQ

type MQ interface {
	// Startup starts the broker.
	Startup() error

	// Shutdown stops the broker.
	Shutdown() error
}

MQ is the broker abstraction.

type MQClient

type MQClient interface {
	// Connect connects to the given broker using the provided conf.
	Connect(*MQClientOptions) error

	// Disconnect breaks connection with the broker.
	Disconnect()
}

MQClient abstracts the broker client logic.

type MQClientOptions

type MQClientOptions struct {
}

MQClientOptions defines options provided to client when connecting.

type MQTTClient

type MQTTClient struct {
	*paho.Client
	// contains filtered or unexported fields
}

MQTTClient provides an MQTT v5 compatible client impl.

func NewMQTTClient

func NewMQTTClient(network string, endpoint string) *MQTTClient

NewMQTTClient creates a client and establish a connection to the given broker.

func (*MQTTClient) Connect

func (c *MQTTClient) Connect(connMsg *paho.Connect) error

Connect send CONNECT msg to the broker on the established underlying network connection.

func (*MQTTClient) Disconnect

func (c *MQTTClient) Disconnect() error

Disconnect sends DISCONNECT msg to broker and always close the network connection

type MQTTClientV3

type MQTTClientV3 struct {
	mqtt.Client
	// contains filtered or unexported fields
}

MQTTClientV3 provides an MQTT v3/v3.1.1 compatible client impl.

func NewMQTTClientV3

func NewMQTTClientV3(network string, endpoint string) *MQTTClientV3

NewMQTTClientV3 creates a client and establish a connection to the given broker.

func (*MQTTClientV3) Connect

func (c *MQTTClientV3) Connect(connMsg *paho.Connect) error

Connect connects to broker using v3.1.1 first and then v3 when the first try failed. NOTE: this method use paho.Connect v5 Connect message to keep compatibility.

func (*MQTTClientV3) Disconnect

func (c *MQTTClientV3) Disconnect() error

Disconnect will end the connection with the server, before waiting 3000 ms to wait for cleaning.

type MQTTServer

type MQTTServer struct {
	*mqtt.Server
}

MQTTServer provides an MQTT v3/v3.1.1 compatible broker impl.

func NewMQTTServer

func NewMQTTServer(name string, endpoint string) *MQTTServer

NewMQTTServer creates a single node MQTT broker with the given name and listen address endpoint.

func (*MQTTServer) Shutdown

func (s *MQTTServer) Shutdown() error

Shutdown stops the broker.

func (*MQTTServer) Startup

func (s *MQTTServer) Startup() error

Startup starts the broker.

type Option

type Option = func(*EmbeddedNats)

func WithAuthorizationToken

func WithAuthorizationToken(t string) Option

func WithHost

func WithHost(host string) Option

WithHost overrides default host "127.0.0.1".

func WithLoggerFile

func WithLoggerFile(file string) Option

WithLoggerFile defines logger for embedded nats, file can be "stdout", "stderr" or any file path. Logging is disabled if file is "", which is the default.

func WithMonitorHost

func WithMonitorHost(host string) Option

WithMonitorHost overrides default host "127.0.0.1".

func WithMonitorPort

func WithMonitorPort(port int) Option

WithMonitorPort overrides default port 8222.

func WithPort

func WithPort(port int) Option

WithPort overrides default port 4222.

func WithRetryCount

func WithRetryCount(c int) Option

WithRetryCount overrides default retry count, 3.

func WithRetryDuration

func WithRetryDuration(d time.Duration) Option

WithRetryDuration overrides default retry duration which is 5 seconds.