proxy

README

gmqtt-proxy-plugin

It's a plugin created for Gmqtt broker, to make it act like a mqtt proxy, with messages buffering support when the destination broker is down.

How to install

1. Clone Gmqtt project

2. Edit plugin_imports.ymlfile in the root directory and add proxy plugin:

packages:
  - admin
  - prometheus
  - federation
  - auth 
  # add proxy plugin in the end of the list
  - proxy 

1. Edit cmd/gmqttd/default_config.yml file and add proxy plugin:

plugin_order:  
  - prometheus
  - admin
  - federation
  # add proxy plugin here
  - proxy

1. Go to plugin directory and clone this project:

git clone https://github.com/Oliveirakun/gmqtt-proxy-plugin.git

1. Rename the directory:

mv gmqtt-proxy-plugin proxy

1. Go back to root directory and install the plugin dependencies:

go get github.com/eclipse/paho.mqtt.golang

1. Export an environment variable with the destination broker uri:

export REMOTE_MQTT_BROKER=tcp://127.0.0.1:1884

1. Run the project:

make run

Routes

Route	Description
/v1/toogle?stop=true	Stop sending messages to destination broker
/v1/toogle?stop=false	Restart sending messages to destination broker
/v1/toogle?stop=false&broker-uri=tcp://127.0.0.1	Restart sending messages to a different broker

LICENSE

Licensed under MIT, see LICENSE

Documentation

Index

• Constants
• Variables
• func New(config config.Config) (server.Plugin, error)
• type Config
  • func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error
  • func (c *Config) Validate() error
• type Consumer
  • func NewConsumer(c chan *gmqtt.Message, s *Semaphore, cl mqtt.Client) *Consumer
  • func (c *Consumer) DequeueMessagesAndSendToBroker()
  • func (c *Consumer) SetClient(cl mqtt.Client)
• type HTTPHandler
  • func NewHTTPHandler(p *Proxy) *HTTPHandler
  • func (h *HTTPHandler) Handle(ctx context.Context, mux *runtime.ServeMux, endpoint string, ...) error
• type Producer
  • func NewProducer(c chan *gmqtt.Message) *Producer
  • func (p *Producer) Enqueue(message *gmqtt.Message)
• type Proxy
  • func (p *Proxy) HookWrapper() server.HookWrapper
  • func (p *Proxy) Load(service server.Server) error
  • func (p *Proxy) Name() string
  • func (p *Proxy) OnMsgArrivedWrapper(pre server.OnMsgArrived) server.OnMsgArrived
  • func (p *Proxy) Unload() error
• type Semaphore
  • func NewSemaphore() *Semaphore
  • func (s *Semaphore) Close()
  • func (s *Semaphore) IsOpen() bool
  • func (s *Semaphore) Open()

Constants

const Name = "proxy"

Variables

var DefaultConfig = Config{}

DefaultConfig is the default configuration.

Functions

func New

func New(config config.Config) (server.Plugin, error)

Types

type Config

type Config struct {
}

Config is the configuration for the proxy plugin.

func (*Config) UnmarshalYAML

func (c *Config) UnmarshalYAML(unmarshal func(interface{}) error) error

func (*Config) Validate

func (c *Config) Validate() error

Validate validates the configuration, and return an error if it is invalid.

type Consumer

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

func NewConsumer

func NewConsumer(c chan *gmqtt.Message, s *Semaphore, cl mqtt.Client) *Consumer

func (*Consumer) DequeueMessagesAndSendToBroker

func (c *Consumer) DequeueMessagesAndSendToBroker()

func (*Consumer) SetClient

func (c *Consumer) SetClient(cl mqtt.Client)

type HTTPHandler

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

func NewHTTPHandler

func NewHTTPHandler(p *Proxy) *HTTPHandler

func (*HTTPHandler) Handle

func (h *HTTPHandler) Handle(ctx context.Context, mux *runtime.ServeMux, endpoint string, opts []grpc.DialOption) error

type Producer

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

func NewProducer

func NewProducer(c chan *gmqtt.Message) *Producer

func (*Producer) Enqueue

func (p *Producer) Enqueue(message *gmqtt.Message)

type Proxy

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

func (*Proxy) HookWrapper

func (p *Proxy) HookWrapper() server.HookWrapper

func (*Proxy) Load

func (p *Proxy) Load(service server.Server) error

func (*Proxy) Name

func (p *Proxy) Name() string

func (*Proxy) OnMsgArrivedWrapper

func (p *Proxy) OnMsgArrivedWrapper(pre server.OnMsgArrived) server.OnMsgArrived

func (*Proxy) Unload

func (p *Proxy) Unload() error

type Semaphore

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

func NewSemaphore

func NewSemaphore() *Semaphore

func (*Semaphore) Close

func (s *Semaphore) Close()

func (*Semaphore) IsOpen

func (s *Semaphore) IsOpen() bool

func (*Semaphore) Open

func (s *Semaphore) Open()