kafka

Documentation

Index

• Constants
• type Bridge
  • func (b *Bridge) ID() string
  • func (b *Bridge) Init(config any) error
  • func (b *Bridge) OnDisconnect(cl *mqtt.Client, err error, expire bool)
  • func (b *Bridge) OnPublished(cl *mqtt.Client, pk packets.Packet)
  • func (b *Bridge) OnSessionEstablished(cl *mqtt.Client, pk packets.Packet)
  • func (b *Bridge) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)
  • func (b *Bridge) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)
  • func (b *Bridge) Provides(bt byte) bool
  • func (b *Bridge) Stop() error
• type Message
  • func (d Message) MarshalBinary() (data []byte, err error)
  • func (d *Message) UnmarshalBinary(data []byte) error
• type Options

Constants

const (
	//Connect mqtt connect
	Connect = "connect"
	//Publish mqtt publish
	Publish = "publish"
	//Subscribe mqtt sub
	Subscribe = "subscribe"
	//Unsubscribe mqtt sub
	Unsubscribe = "unsubscribe"
	//Disconnect mqtt disconenct
	Disconnect = "disconnect"
)

Types

type Bridge

type Bridge struct {
	mqtt.HookBase
	// contains filtered or unexported fields
}

func (*Bridge) ID

func (b *Bridge) ID() string

ID returns the ID of the hook.

func (*Bridge) Init

func (b *Bridge) Init(config any) error

func (*Bridge) OnDisconnect

func (b *Bridge) OnDisconnect(cl *mqtt.Client, err error, expire bool)

OnDisconnect is called when a client is disconnected for any reason.

func (*Bridge) OnPublished

func (b *Bridge) OnPublished(cl *mqtt.Client, pk packets.Packet)

OnPublished is called when a client has published a message to subscribers.

func (*Bridge) OnSessionEstablished

func (b *Bridge) OnSessionEstablished(cl *mqtt.Client, pk packets.Packet)

OnSessionEstablished is called when a new client establishes a session (after OnConnect).

func (*Bridge) OnSubscribed

func (b *Bridge) OnSubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)

OnSubscribed is called when a client subscribes to one or more filters.

func (*Bridge) OnUnsubscribed

func (b *Bridge) OnUnsubscribed(cl *mqtt.Client, pk packets.Packet, reasonCodes []byte, counts []int)

OnUnsubscribed is called when a client unsubscribes from one or more filters.

func (*Bridge) Provides

func (b *Bridge) Provides(bt byte) bool

Provides indicates which hook methods this hook provides.

func (*Bridge) Stop

func (b *Bridge) Stop() error

Stop closes the kafka connection.

type Message

type Message struct {
	Action   string   `json:"action"`
	ClientID string   `json:"clientid"`           // the client id
	Username string   `json:"username"`           // the username of the client
	Remote   string   `json:"remote,omitempty"`   // the remote address of the client
	Listener string   `json:"listener,omitempty"` // the listener the client connected on
	Topics   []string `json:"topics,omitempty"`   // publish topic or subscribe/unsubscribe filters

	Payload         []byte `json:"payload,omitempty"`         // publish payload
	ProtocolVersion byte   `json:"protocolVersion,omitempty"` // mqtt protocol version of the client
	Clean           bool   `json:"clean,omitempty"`           // if the client requested a clean start/session
	Timestamp       int64  `json:"ts"`                        // event time
	// contains filtered or unexported fields
}

Message kafka publish message

func (Message) MarshalBinary

func (d Message) MarshalBinary() (data []byte, err error)

MarshalBinary encodes the values into a json string.

func (*Message) UnmarshalBinary

func (d *Message) UnmarshalBinary(data []byte) error

UnmarshalBinary decodes a json string into a struct.

type Options

type Options struct {
	KafkaOptions *kafkaOptions `json:"kafka-options" yaml:"kafka-options"`
	Rules        rules         `json:"rules" yaml:"rules"`
}