collab/hmq
3
0
Fork 0
mirror of https://github.com/fhmq/hmq.git synced 2026-05-02 14:28:34 +00:00
Code Issues Packages Projects Releases Wiki Activity

Compare commits

base: collab:wpool
Branches Tags
collab:master collab:v5_support collab:revert-102-patch-1 collab:new collab:add_acl collab:plugin_update collab:rpc collab:rpc_1 collab:plugins collab:fix_pool collab:concurrent_access_to_subs collab:restruct collab:context collab:wpool collab:sync_pool collab:pool collab:muti_channel_pool collab:master_code collab:newbroker collab:packet
collab:v1.5.7 collab:v1.5.6 collab:v1.5.5 collab:v1.5.4 collab:v1.5.3 collab:v1.5.2 collab:v1.5.1 collab:v1.5.0 collab:v2.0 collab:v1.4 collab:matser collab:1.3 collab:1.2 collab:1.10 collab:1.01 collab:1.0
..
compare: collab:pool
Branches Tags
collab:master collab:v5_support collab:revert-102-patch-1 collab:new collab:add_acl collab:plugin_update collab:rpc collab:rpc_1 collab:plugins collab:fix_pool collab:concurrent_access_to_subs collab:restruct collab:context collab:wpool collab:sync_pool collab:pool collab:muti_channel_pool collab:master_code collab:newbroker collab:packet
collab:v1.5.7 collab:v1.5.6 collab:v1.5.5 collab:v1.5.4 collab:v1.5.3 collab:v1.5.2 collab:v1.5.1 collab:v1.5.0 collab:v2.0 collab:v1.4 collab:matser collab:1.3 collab:1.2 collab:1.10 collab:1.01 collab:1.0

5 Commits
wpool ... pool

Author SHA1 Message Date
zhouyuyan
f92087eada modify readme 2018-02-02 16:50:56 +08:00
zhouyuyan
a8465585b9 modify usage 2018-02-02 09:54:48 +08:00
zhouyuyan
8337e14452 del buf 2018-02-01 16:57:56 +08:00
zhouyuyan
9dac0e0f1e elastic workerpool 2018-02-01 15:41:16 +08:00
zhouyuyan
4dc8a703a9 add pool 2018-02-01 10:10:54 +08:00
9 changed files with 127 additions and 138 deletions
Expand all files Collapse all files

2
README.md
View File

@@ -28,7 +28,7 @@ Broker Options:
Logging Options: Logging Options:
-d, --debug <bool> Enable debugging output (default false) -d, --debug <bool> Enable debugging output (default false)
-D Debug enabled -D Debug and trace
Cluster Options: Cluster Options:
-r, --router <rurl> Router who maintenance cluster info -r, --router <rurl> Router who maintenance cluster info

21
broker/auth.go
View File

@@ -3,10 +3,13 @@
package broker package broker
import ( import (
"github.com/fhmq/hmq/lib/acl"
"github.com/fsnotify/fsnotify"
"go.uber.org/zap"
"strings" "strings"
"github.com/fhmq/hmq/lib/acl"
"go.uber.org/zap"
"github.com/fsnotify/fsnotify"
) )
const ( const (
@@ -41,10 +44,10 @@ func (b *Broker) handleFsEvent(event fsnotify.Event) error {
case b.config.AclConf: case b.config.AclConf:
if event.Op&fsnotify.Write == fsnotify.Write || if event.Op&fsnotify.Write == fsnotify.Write ||
event.Op&fsnotify.Create == fsnotify.Create { event.Op&fsnotify.Create == fsnotify.Create {
log.Info("text:handling acl config change event:", zap.String("filename", event.Name)) brokerLog.Info("text:handling acl config change event:", zap.String("filename", event.Name))
aclconfig, err := acl.AclConfigLoad(event.Name) aclconfig, err := acl.AclConfigLoad(event.Name)
if err != nil { if err != nil {
log.Error("aclconfig change failed, load acl conf error: ", zap.Error(err)) brokerLog.Error("aclconfig change failed, load acl conf error: ", zap.Error(err))
return err return err
} }
b.AclConfig = aclconfig b.AclConfig = aclconfig
@@ -57,24 +60,24 @@ func (b *Broker) StartAclWatcher() {
go func() { go func() {
wch, e := fsnotify.NewWatcher() wch, e := fsnotify.NewWatcher()
if e != nil { if e != nil {
log.Error("start monitor acl config file error,", zap.Error(e)) brokerLog.Error("start monitor acl config file error,", zap.Error(e))
return return
} }
defer wch.Close() defer wch.Close()
for _, i := range watchList { for _, i := range watchList {
if err := wch.Add(i); err != nil { if err := wch.Add(i); err != nil {
log.Error("start monitor acl config file error,", zap.Error(err)) brokerLog.Error("start monitor acl config file error,", zap.Error(err))
return return
} }
} }
log.Info("watching acl config file change...") brokerLog.Info("watching acl config file change...")
for { for {
select { select {
case evt := <-wch.Events: case evt := <-wch.Events:
b.handleFsEvent(evt) b.handleFsEvent(evt)
case err := <-wch.Errors: case err := <-wch.Errors:
log.Error("error:", zap.Error(err)) brokerLog.Error("error:", zap.Error(err))
} }
} }
}() }()

122
broker/broker.go
View File

@@ -11,17 +11,18 @@ import (
"sync/atomic" "sync/atomic"
"time" "time"
"github.com/eclipse/paho.mqtt.golang/packets"
"github.com/fhmq/hmq/lib/acl" "github.com/fhmq/hmq/lib/acl"
"github.com/fhmq/hmq/pool" "github.com/fhmq/hmq/pool"
"github.com/eclipse/paho.mqtt.golang/packets"
"github.com/shirou/gopsutil/mem" "github.com/shirou/gopsutil/mem"
"go.uber.org/zap" "go.uber.org/zap"
"golang.org/x/net/websocket" "golang.org/x/net/websocket"
) )
const ( var (
MessagePoolNum = 1024 brokerLog *zap.Logger
MessagePoolMessageNum = 1024
) )
type Message struct { type Message struct {
@@ -42,19 +43,10 @@ type Broker struct {
remotes sync.Map remotes sync.Map
nodes map[string]interface{} nodes map[string]interface{}
clusterPool chan *Message clusterPool chan *Message
messagePool chan *Message
sl *Sublist sl *Sublist
rl *RetainList rl *RetainList
queues map[string]int queues map[string]int
// messagePool []chan *Message
}
func newMessagePool() []chan *Message {
pool := make([]chan *Message, 0)
for i := 0; i < MessagePoolNum; i++ {
ch := make(chan *Message, MessagePoolMessageNum)
pool = append(pool, ch)
}
return pool
} }
func NewBroker(config *Config) (*Broker, error) { func NewBroker(config *Config) (*Broker, error) {
@@ -67,12 +59,12 @@ func NewBroker(config *Config) (*Broker, error) {
nodes: make(map[string]interface{}), nodes: make(map[string]interface{}),
queues: make(map[string]int), queues: make(map[string]int),
clusterPool: make(chan *Message), clusterPool: make(chan *Message),
// messagePool: newMessagePool(), messagePool: make(chan *Message),
} }
if b.config.TlsPort != "" { if b.config.TlsPort != "" {
tlsconfig, err := NewTLSConfig(b.config.TlsInfo) tlsconfig, err := NewTLSConfig(b.config.TlsInfo)
if err != nil { if err != nil {
log.Error("new tlsConfig error", zap.Error(err)) brokerLog.Error("new tlsConfig error", zap.Error(err))
return nil, err return nil, err
} }
b.tlsConfig = tlsconfig b.tlsConfig = tlsconfig
@@ -80,7 +72,7 @@ func NewBroker(config *Config) (*Broker, error) {
if b.config.Acl { if b.config.Acl {
aclconfig, err := acl.AclConfigLoad(b.config.AclConf) aclconfig, err := acl.AclConfigLoad(b.config.AclConf)
if err != nil { if err != nil {
log.Error("Load acl conf error", zap.Error(err)) brokerLog.Error("Load acl conf error", zap.Error(err))
return nil, err return nil, err
} }
b.AclConfig = aclconfig b.AclConfig = aclconfig
@@ -89,14 +81,13 @@ func NewBroker(config *Config) (*Broker, error) {
return b, nil return b, nil
} }
func (b *Broker) SubmitWork(msg *Message) { func (b *Broker) StartDispatcher() {
if b.wpool == nil { for {
b.wpool = pool.New(b.config.Worker) msg, ok := <-b.messagePool
} if !ok {
brokerLog.Error("read message from client channel error")
if msg.client.typ == CLUSTER { return
b.clusterPool <- msg }
} else {
b.wpool.Submit(func() { b.wpool.Submit(func() {
ProcessMessage(msg) ProcessMessage(msg)
}) })
@@ -106,9 +97,10 @@ func (b *Broker) SubmitWork(msg *Message) {
func (b *Broker) Start() { func (b *Broker) Start() {
if b == nil { if b == nil {
log.Error("broker is null") brokerLog.Error("broker is null")
return return
} }
go b.StartDispatcher()
//listen clinet over tcp //listen clinet over tcp
if b.config.Port != "" { if b.config.Port != "" {
@@ -157,7 +149,7 @@ func StateMonitor() {
func (b *Broker) StartWebsocketListening() { func (b *Broker) StartWebsocketListening() {
path := b.config.WsPath path := b.config.WsPath
hp := ":" + b.config.WsPort hp := ":" + b.config.WsPort
log.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path)) brokerLog.Info("Start Websocket Listener on:", zap.String("hp", hp), zap.String("path", path))
http.Handle(path, websocket.Handler(b.wsHandler)) http.Handle(path, websocket.Handler(b.wsHandler))
var err error var err error
if b.config.WsTLS { if b.config.WsTLS {
@@ -166,7 +158,7 @@ func (b *Broker) StartWebsocketListening() {
err = http.ListenAndServe(hp, nil) err = http.ListenAndServe(hp, nil)
} }
if err != nil { if err != nil {
log.Error("ListenAndServe:" + err.Error()) brokerLog.Error("ListenAndServe:" + err.Error())
return return
} }
} }
@@ -175,7 +167,7 @@ func (b *Broker) wsHandler(ws *websocket.Conn) {
// io.Copy(ws, ws) // io.Copy(ws, ws)
atomic.AddUint64(&b.cid, 1) atomic.AddUint64(&b.cid, 1)
ws.PayloadType = websocket.BinaryFrame ws.PayloadType = websocket.BinaryFrame
b.handleConnection(CLIENT, ws) b.handleConnection(CLIENT, ws, b.cid)
} }
func (b *Broker) StartClientListening(Tls bool) { func (b *Broker) StartClientListening(Tls bool) {
@@ -185,14 +177,14 @@ func (b *Broker) StartClientListening(Tls bool) {
if Tls { if Tls {
hp = b.config.TlsHost + ":" + b.config.TlsPort hp = b.config.TlsHost + ":" + b.config.TlsPort
l, err = tls.Listen("tcp", hp, b.tlsConfig) l, err = tls.Listen("tcp", hp, b.tlsConfig)
log.Info("Start TLS Listening client on ", zap.String("hp", hp)) brokerLog.Info("Start TLS Listening client on ", zap.String("hp", hp))
} else { } else {
hp := b.config.Host + ":" + b.config.Port hp := b.config.Host + ":" + b.config.Port
l, err = net.Listen("tcp", hp) l, err = net.Listen("tcp", hp)
log.Info("Start Listening client on ", zap.String("hp", hp)) brokerLog.Info("Start Listening client on ", zap.String("hp", hp))
} }
if err != nil { if err != nil {
log.Error("Error listening on ", zap.Error(err)) brokerLog.Error("Error listening on ", zap.Error(err))
return return
} }
tmpDelay := 10 * ACCEPT_MIN_SLEEP tmpDelay := 10 * ACCEPT_MIN_SLEEP
@@ -200,7 +192,7 @@ func (b *Broker) StartClientListening(Tls bool) {
conn, err := l.Accept() conn, err := l.Accept()
if err != nil { if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() { if ne, ok := err.(net.Error); ok && ne.Temporary() {
log.Error("Temporary Client Accept Error(%v), sleeping %dms", brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond)) zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
time.Sleep(tmpDelay) time.Sleep(tmpDelay)
tmpDelay *= 2 tmpDelay *= 2
@@ -208,13 +200,13 @@ func (b *Broker) StartClientListening(Tls bool) {
tmpDelay = ACCEPT_MAX_SLEEP tmpDelay = ACCEPT_MAX_SLEEP
} }
} else { } else {
log.Error("Accept error: %v", zap.Error(err)) brokerLog.Error("Accept error: %v", zap.Error(err))
} }
continue continue
} }
tmpDelay = ACCEPT_MIN_SLEEP tmpDelay = ACCEPT_MIN_SLEEP
atomic.AddUint64(&b.cid, 1) atomic.AddUint64(&b.cid, 1)
go b.handleConnection(CLIENT, conn) go b.handleConnection(CLIENT, conn, b.cid)
} }
} }
@@ -227,7 +219,7 @@ func (b *Broker) Handshake(conn net.Conn) bool {
// Force handshake // Force handshake
if err := nc.Handshake(); err != nil { if err := nc.Handshake(); err != nil {
log.Error("TLS handshake error, ", zap.Error(err)) brokerLog.Error("TLS handshake error, ", zap.Error(err))
return false return false
} }
nc.SetReadDeadline(time.Time{}) nc.SetReadDeadline(time.Time{})
@@ -243,27 +235,28 @@ func TlsTimeout(conn *tls.Conn) {
} }
cs := nc.ConnectionState() cs := nc.ConnectionState()
if !cs.HandshakeComplete { if !cs.HandshakeComplete {
log.Error("TLS handshake timeout") brokerLog.Error("TLS handshake timeout")
nc.Close() nc.Close()
} }
} }
func (b *Broker) StartClusterListening() { func (b *Broker) StartClusterListening() {
var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port
log.Info("Start Listening cluster on ", zap.String("hp", hp)) brokerLog.Info("Start Listening cluster on ", zap.String("hp", hp))
l, e := net.Listen("tcp", hp) l, e := net.Listen("tcp", hp)
if e != nil { if e != nil {
log.Error("Error listening on ", zap.Error(e)) brokerLog.Error("Error listening on ", zap.Error(e))
return return
} }
var idx uint64 = 0
tmpDelay := 10 * ACCEPT_MIN_SLEEP tmpDelay := 10 * ACCEPT_MIN_SLEEP
for { for {
conn, err := l.Accept() conn, err := l.Accept()
if err != nil { if err != nil {
if ne, ok := err.(net.Error); ok && ne.Temporary() { if ne, ok := err.(net.Error); ok && ne.Temporary() {
log.Error("Temporary Client Accept Error(%v), sleeping %dms", brokerLog.Error("Temporary Client Accept Error(%v), sleeping %dms",
zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond)) zap.Error(ne), zap.Duration("sleeping", tmpDelay/time.Millisecond))
time.Sleep(tmpDelay) time.Sleep(tmpDelay)
tmpDelay *= 2 tmpDelay *= 2
@@ -271,30 +264,30 @@ func (b *Broker) StartClusterListening() {
tmpDelay = ACCEPT_MAX_SLEEP tmpDelay = ACCEPT_MAX_SLEEP
} }
} else { } else {
log.Error("Accept error: %v", zap.Error(err)) brokerLog.Error("Accept error: %v", zap.Error(err))
} }
continue continue
} }
tmpDelay = ACCEPT_MIN_SLEEP tmpDelay = ACCEPT_MIN_SLEEP
go b.handleConnection(ROUTER, conn) go b.handleConnection(ROUTER, conn, idx)
} }
} }
func (b *Broker) handleConnection(typ int, conn net.Conn) { func (b *Broker) handleConnection(typ int, conn net.Conn, idx uint64) {
//process connect packet //process connect packet
packet, err := packets.ReadPacket(conn) packet, err := packets.ReadPacket(conn)
if err != nil { if err != nil {
log.Error("read connect packet error: ", zap.Error(err)) brokerLog.Error("read connect packet error: ", zap.Error(err))
return return
} }
if packet == nil { if packet == nil {
log.Error("received nil packet") brokerLog.Error("received nil packet")
return return
} }
msg, ok := packet.(*packets.ConnectPacket) msg, ok := packet.(*packets.ConnectPacket)
if !ok { if !ok {
log.Error("received msg that was not Connect") brokerLog.Error("received msg that was not Connect")
return return
} }
connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket) connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
@@ -302,7 +295,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
connack.SessionPresent = msg.CleanSession connack.SessionPresent = msg.CleanSession
err = connack.Write(conn) err = connack.Write(conn)
if err != nil { if err != nil {
log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier)) brokerLog.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
return return
} }
@@ -342,7 +335,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
case CLIENT: case CLIENT:
old, exist = b.clients.Load(cid) old, exist = b.clients.Load(cid)
if exist { if exist {
log.Warn("client exist, close old...", zap.String("clientID", c.info.clientID)) brokerLog.Warn("client exist, close old...", zap.String("clientID", c.info.clientID))
ol, ok := old.(*client) ol, ok := old.(*client)
if ok { if ok {
ol.Close() ol.Close()
@@ -352,7 +345,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
case ROUTER: case ROUTER:
old, exist = b.routes.Load(cid) old, exist = b.routes.Load(cid)
if exist { if exist {
log.Warn("router exist, close old...") brokerLog.Warn("router exist, close old...")
ol, ok := old.(*client) ol, ok := old.(*client)
if ok { if ok {
ol.Close() ol.Close()
@@ -361,9 +354,7 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
b.routes.Store(cid, c) b.routes.Store(cid, c)
} }
// mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum] c.readLoop(b.messagePool)
c.readLoop()
} }
func (b *Broker) ConnectToDiscovery() { func (b *Broker) ConnectToDiscovery() {
@@ -373,8 +364,8 @@ func (b *Broker) ConnectToDiscovery() {
for { for {
conn, err = net.Dial("tcp", b.config.Router) conn, err = net.Dial("tcp", b.config.Router)
if err != nil { if err != nil {
log.Error("Error trying to connect to route: ", zap.Error(err)) brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
log.Debug("Connect to route timeout ,retry...") brokerLog.Debug("Connect to route timeout ,retry...")
if 0 == tempDelay { if 0 == tempDelay {
tempDelay = 1 * time.Second tempDelay = 1 * time.Second
@@ -390,7 +381,7 @@ func (b *Broker) ConnectToDiscovery() {
} }
break break
} }
log.Debug("connect to router success :", zap.String("Router", b.config.Router)) brokerLog.Debug("connect to router success :", zap.String("Router", b.config.Router))
cid := b.id cid := b.id
info := info{ info := info{
@@ -410,7 +401,7 @@ func (b *Broker) ConnectToDiscovery() {
c.SendConnect() c.SendConnect()
c.SendInfo() c.SendInfo()
go c.readLoop() go c.readLoop(b.clusterPool)
go c.StartPing() go c.StartPing()
} }
@@ -418,7 +409,7 @@ func (b *Broker) processClusterInfo() {
for { for {
msg, ok := <-b.clusterPool msg, ok := <-b.clusterPool
if !ok { if !ok {
log.Error("read message from cluster channel error") brokerLog.Error("read message from cluster channel error")
return return
} }
ProcessMessage(msg) ProcessMessage(msg)
@@ -440,13 +431,13 @@ func (b *Broker) connectRouter(id, addr string) {
conn, err = net.Dial("tcp", addr) conn, err = net.Dial("tcp", addr)
if err != nil { if err != nil {
log.Error("Error trying to connect to route: ", zap.Error(err)) brokerLog.Error("Error trying to connect to route: ", zap.Error(err))
if retryTimes > 50 { if retryTimes > 50 {
return return
} }
log.Debug("Connect to route timeout ,retry...") brokerLog.Debug("Connect to route timeout ,retry...")
if 0 == timeDelay { if 0 == timeDelay {
timeDelay = 1 * time.Second timeDelay = 1 * time.Second
@@ -486,8 +477,7 @@ func (b *Broker) connectRouter(id, addr string) {
c.SendConnect() c.SendConnect()
// mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum] go c.readLoop(b.messagePool)
go c.readLoop()
go c.StartPing() go c.StartPing()
} }
@@ -546,7 +536,7 @@ func (b *Broker) SendLocalSubsToRouter(c *client) {
if len(subInfo.Topics) > 0 { if len(subInfo.Topics) > 0 {
err := c.WriterPacket(subInfo) err := c.WriterPacket(subInfo)
if err != nil { if err != nil {
log.Error("Send localsubs To Router error :", zap.Error(err)) brokerLog.Error("Send localsubs To Router error :", zap.Error(err))
} }
} }
} }
@@ -563,7 +553,7 @@ func (b *Broker) BroadcastInfoMessage(remoteID string, msg *packets.PublishPacke
return true return true
}) })
// log.Info("BroadcastInfoMessage success ") // brokerLog.Info("BroadcastInfoMessage success ")
} }
func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) { func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
@@ -575,7 +565,7 @@ func (b *Broker) BroadcastSubOrUnsubMessage(packet packets.ControlPacket) {
} }
return true return true
}) })
// log.Info("BroadcastSubscribeMessage remotes: ", s.remotes) // brokerLog.Info("BroadcastSubscribeMessage remotes: ", s.remotes)
} }
func (b *Broker) removeClient(c *client) { func (b *Broker) removeClient(c *client) {
@@ -589,7 +579,7 @@ func (b *Broker) removeClient(c *client) {
case REMOTE: case REMOTE:
b.remotes.Delete(clientId) b.remotes.Delete(clientId)
} }
// log.Info("delete client ,", clientId) // brokerLog.Info("delete client ,", clientId)
} }
func (b *Broker) PublishMessage(packet *packets.PublishPacket) { func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
@@ -603,7 +593,7 @@ func (b *Broker) PublishMessage(packet *packets.PublishPacket) {
if sub != nil { if sub != nil {
err := sub.client.WriterPacket(packet) err := sub.client.WriterPacket(packet)
if err != nil { if err != nil {
log.Error("process message for psub error, ", zap.Error(err)) brokerLog.Error("process message for psub error, ", zap.Error(err))
} }
} }
} }

60
broker/client.go
View File

@@ -86,11 +86,8 @@ func (c *client) init() {
c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0] c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0]
} }
func (c *client) keepAlive(ch chan int) { func (c *client) keepAlive(ch chan int, mpool chan *Message) {
defer close(ch) defer close(ch)
b := c.broker
keepalive := time.Duration(c.info.keepalive*3/2) * time.Second keepalive := time.Duration(c.info.keepalive*3/2) * time.Second
timer := time.NewTimer(keepalive) timer := time.NewTimer(keepalive)
@@ -103,11 +100,9 @@ func (c *client) keepAlive(ch chan int) {
timer.Reset(keepalive) timer.Reset(keepalive)
continue continue
} }
log.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive)) brokerLog.Error("Client exceeded timeout, disconnecting. ", zap.String("ClientID", c.info.clientID), zap.Uint16("keepalive", c.info.keepalive))
msg := &Message{client: c, packet: DisconnectdPacket} msg := &Message{client: c, packet: DisconnectdPacket}
b.SubmitWork(msg) mpool <- msg
timer.Stop() timer.Stop()
return return
case _, ok := <-c.closed: case _, ok := <-c.closed:
@@ -118,20 +113,19 @@ func (c *client) keepAlive(ch chan int) {
} }
} }
func (c *client) readLoop() { func (c *client) readLoop(mpool chan *Message) {
nc := c.conn nc := c.conn
b := c.broker if nc == nil || mpool == nil {
if nc == nil || b == nil {
return return
} }
ch := make(chan int, 1000) ch := make(chan int, 1000)
go c.keepAlive(ch) go c.keepAlive(ch, mpool)
for { for {
packet, err := packets.ReadPacket(nc) packet, err := packets.ReadPacket(nc)
if err != nil { if err != nil {
log.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
break break
} }
// keepalive channel // keepalive channel
@@ -141,11 +135,11 @@ func (c *client) readLoop() {
client: c, client: c,
packet: packet, packet: packet,
} }
b.SubmitWork(msg) mpool <- msg
} }
msg := &Message{client: c, packet: DisconnectdPacket} msg := &Message{client: c, packet: DisconnectdPacket}
b.SubmitWork(msg) mpool <- msg
} }
func ProcessMessage(msg *Message) { func ProcessMessage(msg *Message) {
@@ -155,7 +149,7 @@ func ProcessMessage(msg *Message) {
return return
} }
log.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID)) brokerLog.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID))
switch ca.(type) { switch ca.(type) {
case *packets.ConnackPacket: case *packets.ConnackPacket:
case *packets.ConnectPacket: case *packets.ConnectPacket:
@@ -180,7 +174,7 @@ func ProcessMessage(msg *Message) {
case *packets.DisconnectPacket: case *packets.DisconnectPacket:
c.Close() c.Close()
default: default:
log.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID)) brokerLog.Info("Recv Unknow message.......", zap.String("ClientID", c.info.clientID))
} }
} }
@@ -196,7 +190,7 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
} }
if !c.CheckTopicAuth(PUB, topic) { if !c.CheckTopicAuth(PUB, topic) {
log.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) brokerLog.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
return return
} }
@@ -207,21 +201,21 @@ func (c *client) ProcessPublish(packet *packets.PublishPacket) {
puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket) puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
puback.MessageID = packet.MessageID puback.MessageID = packet.MessageID
if err := c.WriterPacket(puback); err != nil { if err := c.WriterPacket(puback); err != nil {
log.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
return return
} }
c.ProcessPublishMessage(packet) c.ProcessPublishMessage(packet)
case QosExactlyOnce: case QosExactlyOnce:
return return
default: default:
log.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID)) brokerLog.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID))
return return
} }
if packet.Retain { if packet.Retain {
if b := c.broker; b != nil { if b := c.broker; b != nil {
err := b.rl.Insert(topic, packet) err := b.rl.Insert(topic, packet)
if err != nil { if err != nil {
log.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("Insert Retain Message error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
} }
} }
} }
@@ -241,7 +235,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
topic := packet.TopicName topic := packet.TopicName
r := b.sl.Match(topic) r := b.sl.Match(topic)
// log.Info("psubs num: ", len(r.psubs)) // brokerLog.Info("psubs num: ", len(r.psubs))
if len(r.qsubs) == 0 && len(r.psubs) == 0 { if len(r.qsubs) == 0 && len(r.psubs) == 0 {
return return
} }
@@ -255,7 +249,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
if sub != nil { if sub != nil {
err := sub.client.WriterPacket(packet) err := sub.client.WriterPacket(packet)
if err != nil { if err != nil {
log.Error("process message for psub error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("process message for psub error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
} }
} }
} }
@@ -265,7 +259,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
t := "$queue/" + topic t := "$queue/" + topic
cnt, exist := b.queues[t] cnt, exist := b.queues[t]
if exist { if exist {
// log.Info("queue index : ", cnt) // brokerLog.Info("queue index : ", cnt)
for _, sub := range r.qsubs { for _, sub := range r.qsubs {
if sub.client.typ == ROUTER { if sub.client.typ == ROUTER {
if typ != CLIENT { if typ != CLIENT {
@@ -281,7 +275,7 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
if sub != nil { if sub != nil {
err := sub.client.WriterPacket(packet) err := sub.client.WriterPacket(packet)
if err != nil { if err != nil {
log.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("send publish error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
} }
} }
@@ -335,7 +329,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
t := topic t := topic
//check topic auth for client //check topic auth for client
if !c.CheckTopicAuth(SUB, topic) { if !c.CheckTopicAuth(SUB, topic) {
log.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID)) brokerLog.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
retcodes = append(retcodes, QosFailure) retcodes = append(retcodes, QosFailure)
continue continue
} }
@@ -382,7 +376,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
} }
err := b.sl.Insert(sub) err := b.sl.Insert(sub)
if err != nil { if err != nil {
log.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("Insert subscription error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
retcodes = append(retcodes, QosFailure) retcodes = append(retcodes, QosFailure)
} else { } else {
retcodes = append(retcodes, qoss[i]) retcodes = append(retcodes, qoss[i])
@@ -392,7 +386,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
err := c.WriterPacket(suback) err := c.WriterPacket(suback)
if err != nil { if err != nil {
log.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
return return
} }
//broadcast subscribe message //broadcast subscribe message
@@ -404,7 +398,7 @@ func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
for _, t := range topics { for _, t := range topics {
packets := b.rl.Match(t) packets := b.rl.Match(t)
for _, packet := range packets { for _, packet := range packets {
log.Info("process retain message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID)) brokerLog.Info("process retain message: ", zap.Any("packet", packet), zap.String("ClientID", c.info.clientID))
if packet != nil { if packet != nil {
c.WriterPacket(packet) c.WriterPacket(packet)
} }
@@ -451,7 +445,7 @@ func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
err := c.WriterPacket(unsuback) err := c.WriterPacket(unsuback)
if err != nil { if err != nil {
log.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
return return
} }
// //process ubsubscribe message // //process ubsubscribe message
@@ -480,7 +474,7 @@ func (c *client) ProcessPing() {
resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket) resp := packets.NewControlPacket(packets.Pingresp).(*packets.PingrespPacket)
err := c.WriterPacket(resp) err := c.WriterPacket(resp)
if err != nil { if err != nil {
log.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("send PingResponse error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
return return
} }
} }
@@ -511,7 +505,7 @@ func (c *client) Close() {
for _, sub := range subs { for _, sub := range subs {
err := b.sl.Remove(sub) err := b.sl.Remove(sub)
if err != nil { if err != nil {
log.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID)) brokerLog.Error("closed client but remove sublist error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
} }
} }
if c.typ == CLIENT { if c.typ == CLIENT {
@@ -533,7 +527,7 @@ func (c *client) Close() {
} }
func (c *client) WriterPacket(packet packets.ControlPacket) error { func (c *client) WriterPacket(packet packets.ControlPacket) error {
if c == nil || packet == nil { if packet == nil {
return nil return nil
} }

19
broker/config.go
View File

@@ -9,10 +9,11 @@ import (
"errors" "errors"
"flag" "flag"
"fmt" "fmt"
"github.com/fhmq/hmq/logger"
"go.uber.org/zap"
"io/ioutil" "io/ioutil"
"os" "os"
"github.com/fhmq/hmq/logger"
"go.uber.org/zap"
) )
type Config struct { type Config struct {
@@ -51,10 +52,6 @@ var DefaultConfig *Config = &Config{
Acl: false, Acl: false,
} }
var (
log *zap.Logger
)
func showHelp() { func showHelp() {
fmt.Printf("%s\n", usageStr) fmt.Printf("%s\n", usageStr)
os.Exit(0) os.Exit(0)
@@ -108,7 +105,7 @@ func ConfigureConfig(args []string) (*Config, error) {
}) })
logger.InitLogger(config.Debug) logger.InitLogger(config.Debug)
log = logger.Get().Named("Broker") brokerLog = logger.Get().Named("Broker")
if configFile != "" { if configFile != "" {
tmpConfig, e := LoadConfig(configFile) tmpConfig, e := LoadConfig(configFile)
@@ -131,15 +128,15 @@ func LoadConfig(filename string) (*Config, error) {
content, err := ioutil.ReadFile(filename) content, err := ioutil.ReadFile(filename)
if err != nil { if err != nil {
log.Error("Read config file error: ", zap.Error(err)) brokerLog.Error("Read config file error: ", zap.Error(err))
return nil, err return nil, err
} }
// log.Info(string(content)) // brokerLog.Info(string(content))
var config Config var config Config
err = json.Unmarshal(content, &config) err = json.Unmarshal(content, &config)
if err != nil { if err != nil {
log.Error("Unmarshal config file error: ", zap.Error(err)) brokerLog.Error("Unmarshal config file error: ", zap.Error(err))
return nil, err return nil, err
} }
@@ -171,7 +168,7 @@ func (config *Config) check() error {
if config.TlsPort != "" { if config.TlsPort != "" {
if config.TlsInfo.CertFile == "" || config.TlsInfo.KeyFile == "" { if config.TlsInfo.CertFile == "" || config.TlsInfo.KeyFile == "" {
log.Error("tls config error, no cert or key file.") brokerLog.Error("tls config error, no cert or key file.")
return errors.New("tls config error, no cert or key file.") return errors.New("tls config error, no cert or key file.")
} }
if config.TlsHost == "" { if config.TlsHost == "" {

22
broker/info.go
View File

@@ -4,10 +4,12 @@ package broker
import ( import (
"fmt" "fmt"
simplejson "github.com/bitly/go-simplejson" "time"
"github.com/eclipse/paho.mqtt.golang/packets" "github.com/eclipse/paho.mqtt.golang/packets"
"go.uber.org/zap" "go.uber.org/zap"
"time"
simplejson "github.com/bitly/go-simplejson"
) )
func (c *client) SendInfo() { func (c *client) SendInfo() {
@@ -19,7 +21,7 @@ func (c *client) SendInfo() {
infoMsg := NewInfo(c.broker.id, url, false) infoMsg := NewInfo(c.broker.id, url, false)
err := c.WriterPacket(infoMsg) err := c.WriterPacket(infoMsg)
if err != nil { if err != nil {
log.Error("send info message error, ", zap.Error(err)) brokerLog.Error("send info message error, ", zap.Error(err))
return return
} }
} }
@@ -32,7 +34,7 @@ func (c *client) StartPing() {
case <-timeTicker.C: case <-timeTicker.C:
err := c.WriterPacket(ping) err := c.WriterPacket(ping)
if err != nil { if err != nil {
log.Error("ping error: ", zap.Error(err)) brokerLog.Error("ping error: ", zap.Error(err))
c.Close() c.Close()
} }
case _, ok := <-c.closed: case _, ok := <-c.closed:
@@ -55,10 +57,10 @@ func (c *client) SendConnect() {
m.Keepalive = uint16(60) m.Keepalive = uint16(60)
err := c.WriterPacket(m) err := c.WriterPacket(m)
if err != nil { if err != nil {
log.Error("send connect message error, ", zap.Error(err)) brokerLog.Error("send connect message error, ", zap.Error(err))
return return
} }
log.Info("send connect success") brokerLog.Info("send connect success")
} }
func NewInfo(sid, url string, isforword bool) *packets.PublishPacket { func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
@@ -67,7 +69,7 @@ func NewInfo(sid, url string, isforword bool) *packets.PublishPacket {
pub.TopicName = BrokerInfoTopic pub.TopicName = BrokerInfoTopic
pub.Retain = false pub.Retain = false
info := fmt.Sprintf(`{"brokerID":"%s","brokerUrl":"%s"}`, sid, url) info := fmt.Sprintf(`{"brokerID":"%s","brokerUrl":"%s"}`, sid, url)
// log.Info("new info", string(info)) // brokerLog.Info("new info", string(info))
pub.Payload = []byte(info) pub.Payload = []byte(info)
return pub return pub
} }
@@ -79,17 +81,17 @@ func (c *client) ProcessInfo(packet *packets.PublishPacket) {
return return
} }
log.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload))) brokerLog.Info("recv remoteInfo: ", zap.String("payload", string(packet.Payload)))
js, err := simplejson.NewJson(packet.Payload) js, err := simplejson.NewJson(packet.Payload)
if err != nil { if err != nil {
log.Warn("parse info message err", zap.Error(err)) brokerLog.Warn("parse info message err", zap.Error(err))
return return
} }
routes, err := js.Get("data").Map() routes, err := js.Get("data").Map()
if routes == nil { if routes == nil {
log.Error("receive info message error, ", zap.Error(err)) brokerLog.Error("receive info message error, ", zap.Error(err))
return return
} }

11
broker/retain.go
View File

@@ -1,8 +1,9 @@
package broker package broker
import ( import (
"github.com/eclipse/paho.mqtt.golang/packets"
"sync" "sync"
"github.com/eclipse/paho.mqtt.golang/packets"
) )
type RetainList struct { type RetainList struct {
@@ -38,7 +39,7 @@ func (r *RetainList) Insert(topic string, buf *packets.PublishPacket) error {
if err != nil { if err != nil {
return err return err
} }
// log.Info("insert tokens:", tokens) // brokerLog.Info("insert tokens:", tokens)
r.Lock() r.Lock()
l := r.root l := r.root
@@ -71,7 +72,7 @@ func (r *RetainList) Match(topic string) []*packets.PublishPacket {
l := r.root l := r.root
matchRLevel(l, tokens, results) matchRLevel(l, tokens, results)
r.Unlock() r.Unlock()
// log.Info("results: ", results) // brokerLog.Info("results: ", results)
return results.msg return results.msg
} }
@@ -81,7 +82,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) {
if l == nil { if l == nil {
return return
} }
// log.Info("l info :", l.nodes) // brokerLog.Info("l info :", l.nodes)
if t == "#" { if t == "#" {
for _, n := range l.nodes { for _, n := range l.nodes {
n.GetAll(results) n.GetAll(results)
@@ -110,7 +111,7 @@ func matchRLevel(l *rlevel, toks []string, results *RetainResult) {
} }
func (r *rnode) GetAll(results *RetainResult) { func (r *rnode) GetAll(results *RetainResult) {
// log.Info("node 's message: ", string(r.msg)) // brokerLog.Info("node 's message: ", string(r.msg))
if r.msg != nil { if r.msg != nil {
results.msg = append(results.msg, r.msg) results.msg = append(results.msg, r.msg)
} }

5
broker/sublist.go
View File

@@ -4,8 +4,9 @@ package broker
import ( import (
"errors" "errors"
"go.uber.org/zap"
"sync" "sync"
"go.uber.org/zap"
) )
// A result structure better optimized for queue subs. // A result structure better optimized for queue subs.
@@ -210,7 +211,7 @@ func (s *Sublist) Match(topic string) *SublistResult {
tokens, err := PublishTopicCheckAndSpilt(topic) tokens, err := PublishTopicCheckAndSpilt(topic)
if err != nil { if err != nil {
log.Error("\tserver/sublist.go: ", zap.Error(err)) brokerLog.Error("\tserver/sublist.go: ", zap.Error(err))
return nil return nil
} }

3
main.go
View File

@@ -8,10 +8,11 @@ package main
import ( import (
"fmt" "fmt"
"github.com/fhmq/hmq/broker"
"os" "os"
"os/signal" "os/signal"
"runtime" "runtime"
"github.com/fhmq/hmq/broker"
) )
func main() { func main() {