clients

Documentation

Index

• Variables
• type Client
  • func NewClient(c net.Conn, r *circ.Reader, w *circ.Writer, s *system.Info) *Client
  • func NewClientStub(s *system.Info) *Client
  • func (cl *Client) ClearBuffers()
  • func (cl *Client) ForgetSubscription(filter string)
  • func (cl *Client) Identify(lid string, pk packets.Packet, ac auth.Controller)
  • func (cl *Client) Info() events.Client
  • func (cl *Client) NextPacketID() uint32
  • func (cl *Client) NoteSubscription(filter string, qos byte)
  • func (cl *Client) Read(packetHandler func(*Client, packets.Packet) error) error
  • func (cl *Client) ReadFixedHeader(fh *packets.FixedHeader) error
  • func (cl *Client) ReadPacket(fh *packets.FixedHeader) (pk packets.Packet, err error)
  • func (cl *Client) Start()
  • func (cl *Client) Stop(err error)
  • func (cl *Client) StopCause() error
  • func (cl *Client) WritePacket(pk packets.Packet) (n int, err error)
• type Clients
  • func New() *Clients
  • func (cl *Clients) Add(val *Client)
  • func (cl *Clients) Delete(id string)
  • func (cl *Clients) Get(id string) (*Client, bool)
  • func (cl *Clients) GetAll() map[string]*Client
  • func (cl *Clients) GetByListener(id string) []*Client
  • func (cl *Clients) Len() int
• type Inflight
  • func (i *Inflight) ClearExpired(expiry int64) int64
  • func (i *Inflight) Delete(key uint16) bool
  • func (i *Inflight) Get(key uint16) (InflightMessage, bool)
  • func (i *Inflight) GetAll() map[uint16]InflightMessage
  • func (i *Inflight) Len() int
  • func (i *Inflight) Set(key uint16, in InflightMessage) bool
• type InflightMessage
• type LWT
• type State

Variables

var (

	// ErrConnectionClosed is returned when operating on a closed
	// connection and/or when no error cause has been given.
	ErrConnectionClosed = errors.New("connection not open")
)

Types

type Client

type Client struct {
	State        State           // 客户端的运行状态.
	LWT          LWT             // 遗言.
	Inflight     *Inflight       // 正在处理的消息.
	sync.RWMutex                 // 读写锁
	Username     []byte          // 用户名.
	AC           auth.Controller // 授权控制器
	Listener     string          // websocket 还是 tcp的.
	ID           string          // 客户端唯一标识.

	R             *circ.Reader         // 入流.
	W             *circ.Writer         // 出流.
	Subscriptions topics.Subscriptions // 订阅列表.

	CleanSession bool // 是否清楚会话
	// contains filtered or unexported fields
}

客户端的状态

func NewClient

func NewClient(c net.Conn, r *circ.Reader, w *circ.Writer, s *system.Info) *Client

NewClient returns a new instance of Client.

func NewClientStub

func NewClientStub(s *system.Info) *Client

NewClientStub returns an instance of Client with basic initializations. This method is typically called by the persistence restoration system.

func (*Client) ClearBuffers

func (cl *Client) ClearBuffers()

ClearBuffers sets the read/write buffers to nil so they can be deallocated automatically when no longer in use.

func (*Client) ForgetSubscription

func (cl *Client) ForgetSubscription(filter string)

ForgetSubscription forgests a subscription note for the client.

func (*Client) Identify

func (cl *Client) Identify(lid string, pk packets.Packet, ac auth.Controller)

Identify sets the identification values of a client instance.

func (*Client) Info

func (cl *Client) Info() events.Client

Info returns an event-version of a client, containing minimal information.

func (*Client) NextPacketID

func (cl *Client) NextPacketID() uint32

NextPacketID returns the next packet id for a client, looping back to 0 if the maximum ID has been reached.

func (*Client) NoteSubscription

func (cl *Client) NoteSubscription(filter string, qos byte)

NoteSubscription makes a note of a subscription for the client.

func (*Client) Read

func (cl *Client) Read(packetHandler func(*Client, packets.Packet) error) error

Read loops forever reading new packets from a client connection until an error is encountered (or the connection is closed).

func (*Client) ReadFixedHeader

func (cl *Client) ReadFixedHeader(fh *packets.FixedHeader) error

ReadFixedHeader reads in the values of the next packet's fixed header.

func (*Client) ReadPacket

func (cl *Client) ReadPacket(fh *packets.FixedHeader) (pk packets.Packet, err error)

ReadPacket 将剩余的缓冲区读取到 MQTT 数据包中。

func (*Client) Start

func (cl *Client) Start()

Start begins the client goroutines reading and writing packets.

func (*Client) Stop

func (cl *Client) Stop(err error)

Stop instructs the client to shut down all processing goroutines and disconnect. A cause error may be passed to identfy the reason for stopping.

func (*Client) StopCause

func (cl *Client) StopCause() error

StopCause returns the reason the client connection was stopped, if any.

func (*Client) WritePacket

func (cl *Client) WritePacket(pk packets.Packet) (n int, err error)

组包发给客户端

type Clients

type Clients struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Clients contains a map of the clients known by the broker.

func New

func New() *Clients

New returns an instance of Clients.

func (*Clients) Add

func (cl *Clients) Add(val *Client)

Add adds a new client to the clients map, keyed on client id.

func (*Clients) Delete

func (cl *Clients) Delete(id string)

Delete removes a client from the internal map.

func (*Clients) Get

func (cl *Clients) Get(id string) (*Client, bool)

Get returns the value of a client if it exists.

func (*Clients) GetAll

func (cl *Clients) GetAll() map[string]*Client

GetAll returns all the clients.

func (*Clients) GetByListener

func (cl *Clients) GetByListener(id string) []*Client

GetByListener returns clients matching a listener id.

func (*Clients) Len

func (cl *Clients) Len() int

Len returns the length of the clients map.

type Inflight

type Inflight struct {
	sync.RWMutex
	// contains filtered or unexported fields
}

Inflight is a map of InflightMessage keyed on packet id.

func (*Inflight) ClearExpired

func (i *Inflight) ClearExpired(expiry int64) int64

删除过期消息

func (*Inflight) Delete

func (i *Inflight) Delete(key uint16) bool

删除指定消息

func (*Inflight) Get

func (i *Inflight) Get(key uint16) (InflightMessage, bool)

Get returns the value of an in-flight message if it exists.

func (*Inflight) GetAll

func (i *Inflight) GetAll() map[uint16]InflightMessage

获取全部正确处理的消息

func (*Inflight) Len

func (i *Inflight) Len() int

消息队列长度

func (*Inflight) Set

func (i *Inflight) Set(key uint16, in InflightMessage) bool

Set stores the packet of an Inflight message, keyed on message id. Returns true if the inflight message was new.

type InflightMessage

type InflightMessage struct {
	Packet  packets.Packet // 正在处理的消息.
	Sent    int64          // 上次重发时间.
	Created int64          // 消息创建时间
	Resends int            // 消息被重新发送了多少次.
}

InflightMessage contains data about a packet which is currently in-flight.

type LWT

type LWT struct {
	Message []byte // 消息
	Topic   string // 主题
	Qos     byte   // 发送质量
	Retain  bool   // 是否保留
}

遗言

type State

type State struct {
	Done uint32 // 判断是否断开连接.
	// contains filtered or unexported fields
}

客户端的状态.