remove subscriptions when a client disconnect

simple fix
2026-04-30 05:28:34 +00:00 · 2019-01-18 14:38:51 +01:00 · 2019-01-03 21:23:14 +01:00
43 changed files with 640 additions and 1650 deletions
--- a/.vscode/settings.json
+++ b/.vscode/settings.json
@@ -1,8 +0,0 @@
-{
-    "go.lintFlags": [
-        "--disable=all",
-        "--enable=errcheck,varcheck,deadcode",
-        "--enable=varcheck",
-        "--enable=deadcode"
-    ]
-}
--- a/15
+++ b/15
@@ -1,12 +1,11 @@
-FROM golang:1.12 as builder
-WORKDIR /go/src/github.com/fhmq/hmq
-COPY . .
-RUN CGO_ENABLED=0 go build -o hmq -a -ldflags '-extldflags "-static"' .
+FROM alpine
+COPY hmq /
+COPY ssl /ssl
+COPY conf /conf

-
-FROM alpine:3.8
-WORKDIR /
-COPY --from=builder /go/src/github.com/fhmq/hmq/hmq .
 EXPOSE 1883
+EXPOSE 1888
+EXPOSE 8883
+EXPOSE 1993

 CMD ["/hmq"]
--- a/README.md
+++ b/README.md
@@ -5,6 +5,8 @@ Free and High Performance MQTT Broker
 Golang MQTT Broker, Version 3.1.1, and Compatible
 for [eclipse paho client](https://github.com/eclipse?utf8=%E2%9C%93&q=mqtt&type=&language=) and mosquitto-client

+Download: [click here](https://github.com/fhmq/hmq/releases)
+
 ## RUNNING
 ```bash
 $ go get github.com/fhmq/hmq
@@ -58,10 +60,8 @@ Common Options:
 		"certFile": "tls/server/cert.pem",
 		"keyFile": "tls/server/key.pem"
 	},
-	"plugins": {
-		"auth": "authhttp",
-		"bridge": "kafka"
-	}
+	"acl":true,
+	"aclConf":"conf/acl.conf"
 }
 ~~~

@@ -81,24 +81,7 @@ Common Options:

 * TLS/SSL Support

-* Auth Support
-	* Auth Connect
-	* Auth ACL
-	* Cache Support
-
-* Kafka Bridge Support
-	* Action Deliver
-	* Regexp Deliver
-
-* HTTP API
-	* Disconnect Connect (future more)
-
-### Share SUBSCRIBE
-~~~
-| Prefix              | Examples                                  | Publish                      |
-| ------------------- |-------------------------------------------|--------------------------- --|
-| $share/<group>/topic  | mosquitto_sub -t ‘$share/<group>/topic’ | mosquitto_pub -t ‘topic’     |
-~~~
+* Flexible  ACL

 ### Cluster
 ```bash
@@ -109,7 +92,58 @@ Common Options:
 2, config router in hmq.config  ("router": "127.0.0.1:9888")
 
 ```
-Other Version Of Cluster Based On gRPC: [click here](https://github.com/fhmq/rhmq)
+
+### ACL Configure
+#### The ACL rules define:
+~~~
+Allow | type | value | pubsub | Topics
+~~~
+#### ACL Config
+~~~
+## type clientid , username, ipaddr
+##pub 1 ,  sub 2,  pubsub 3
+## %c is clientid , %u is username
+allow      ip          127.0.0.1   2     $SYS/#
+allow      clientid    0001        3     #
+allow      username    admin       3     #
+allow      username    joy         3     /test,hello/world 
+allow      clientid    *           1     toCloud/%c
+allow      username    *           1     toCloud/%u
+deny       clientid    *           3     #
+~~~
+
+~~~
+#allow local sub $SYS topic
+allow      ip          127.0.0.1   2    $SYS/#
+~~~
+~~~
+#allow client who's id with 0001 or username with admin pub sub all topic
+allow      clientid    0001        3        #
+allow      username    admin       3        #
+~~~
+~~~
+#allow client with the username joy can pub sub topic '/test' and 'hello/world'
+allow      username    joy         3     /test,hello/world 
+~~~
+~~~
+#allow all client pub the topic toCloud/{clientid/username}
+allow      clientid    *         1         toCloud/%c
+allow      username    *         1         toCloud/%u
+~~~
+~~~
+#deny all client pub sub all topic
+deny       clientid    *         3           #
+~~~
+Client match acl rule one by one
+~~~
+          ---------              ---------              ---------
+Client -> | Rule1 | --nomatch--> | Rule2 | --nomatch--> | Rule3 | --> 
+          ---------              ---------              ---------
+              |                      |                      |
+            match                  match                  match
+             \|/                    \|/                    \|/
+        allow | deny           allow | deny           allow | deny
+~~~

 ### Online/Offline Notification
 ```bash
@@ -135,9 +169,4 @@ Other Version Of Cluster Based On gRPC: [click here](https://github.com/fhmq/rhm

 ## Reference

-* Surgermq.(https://github.com/surgemq/surgemq)
-
-## Benchmark Tool
-
-* https://github.com/inovex/mqtt-stresser
-* https://github.com/krylovsk/mqtt-benchmark
+* Surgermq.(https://github.com/surgemq/surgemq)
--- a/broker/auth.go
+++ b/broker/auth.go
@@ -1,40 +1,81 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
+	"github.com/fhmq/hmq/lib/acl"
+	"github.com/fsnotify/fsnotify"
+	"go.uber.org/zap"
 	"strings"
 )

 const (
-	SUB = "1"
-	PUB = "2"
+	PUB = 1
+	SUB = 2
 )

-func (b *Broker) CheckTopicAuth(action, clientID, username, ip, topic string) bool {
-	if b.auth != nil {
-		if strings.HasPrefix(topic, "$SYS/broker/connection/clients/") {
-			return true
+func (c *client) CheckTopicAuth(typ int, topic string) bool {
+	if c.typ != CLIENT || !c.broker.config.Acl {
+		return true
+	}
+	if strings.HasPrefix(topic, "$queue/") {
+		topic = string([]byte(topic)[7:])
+		if topic == "" {
+			return false
 		}
+	}
+	ip := c.info.remoteIP
+	username := string(c.info.username)
+	clientid := string(c.info.clientID)
+	aclInfo := c.broker.AclConfig
+	return acl.CheckTopicAuth(aclInfo, typ, ip, username, clientid, topic)

-		if strings.HasPrefix(topic, "$share/") && action == SUB {
-			substr := groupCompile.FindStringSubmatch(topic)
-			if len(substr) != 3 {
-				return false
+}
+
+var (
+	watchList = []string{"./conf"}
+)
+
+func (b *Broker) handleFsEvent(event fsnotify.Event) error {
+	switch event.Name {
+	case b.config.AclConf:
+		if event.Op&fsnotify.Write == fsnotify.Write ||
+			event.Op&fsnotify.Create == fsnotify.Create {
+			log.Info("text:handling acl config change event:", zap.String("filename", event.Name))
+			aclconfig, err := acl.AclConfigLoad(event.Name)
+			if err != nil {
+				log.Error("aclconfig change failed, load acl conf error: ", zap.Error(err))
+				return err
 			}
-			topic = substr[2]
+			b.AclConfig = aclconfig
 		}
-
-		return b.auth.CheckACL(action, clientID, username, ip, topic)
 	}
-
-	return true
-
+	return nil
 }

-func (b *Broker) CheckConnectAuth(clientID, username, password string) bool {
-	if b.auth != nil {
-		return b.auth.CheckConnect(clientID, username, password)
-	}
-
-	return true
+func (b *Broker) StartAclWatcher() {
+	go func() {
+		wch, e := fsnotify.NewWatcher()
+		if e != nil {
+			log.Error("start monitor acl config file error,", zap.Error(e))
+			return
+		}
+		defer wch.Close()

+		for _, i := range watchList {
+			if err := wch.Add(i); err != nil {
+				log.Error("start monitor acl config file error,", zap.Error(err))
+				return
+			}
+		}
+		log.Info("watching acl config file change...")
+		for {
+			select {
+			case evt := <-wch.Events:
+				b.handleFsEvent(evt)
+			case err := <-wch.Errors:
+				log.Error("error:", zap.Error(err))
+			}
+		}
+	}()
 }
--- a/broker/bridge.go
+++ b/broker/bridge.go
@@ -1,15 +0,0 @@
-package broker
-
-import (
-	"github.com/fhmq/hmq/plugins/bridge"
-	"go.uber.org/zap"
-)
-
-func (b *Broker) Publish(e *bridge.Elements) {
-	if b.bridgeMQ != nil {
-		err := b.bridgeMQ.Publish(e)
-		if err != nil {
-			log.Error("send message to mq error.", zap.Error(err))
-		}
-	}
-}
--- a/broker/broker.go
+++ b/broker/broker.go
@@ -1,3 +1,5 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
@@ -5,18 +7,17 @@ import (
 	"fmt"
 	"net"
 	"net/http"
+	"runtime/debug"
 	"sync"
+	"sync/atomic"
 	"time"

-	"github.com/fhmq/hmq/plugins/bridge"
-
-	"github.com/fhmq/hmq/plugins/auth"
-
-	"github.com/fhmq/hmq/broker/lib/sessions"
-	"github.com/fhmq/hmq/broker/lib/topics"
-
 	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/fhmq/hmq/lib/acl"
+	"github.com/fhmq/hmq/lib/sessions"
+	"github.com/fhmq/hmq/lib/topics"
 	"github.com/fhmq/hmq/pool"
+	"github.com/shirou/gopsutil/mem"
 	"go.uber.org/zap"
 	"golang.org/x/net/websocket"
 )
@@ -33,19 +34,21 @@ type Message struct {

 type Broker struct {
 	id          string
+	cid         uint64
 	mu          sync.Mutex
 	config      *Config
 	tlsConfig   *tls.Config
+	AclConfig   *acl.ACLConfig
 	wpool       *pool.WorkerPool
 	clients     sync.Map
 	routes      sync.Map
 	remotes     sync.Map
 	nodes       map[string]interface{}
 	clusterPool chan *Message
+	queues      map[string]int
 	topicsMgr   *topics.Manager
 	sessionMgr  *sessions.Manager
-	auth        auth.Auth
-	bridgeMQ    bridge.BridgeMQ
+	// messagePool []chan *Message
 }

 func newMessagePool() []chan *Message {
@@ -58,15 +61,12 @@ func newMessagePool() []chan *Message {
 }

 func NewBroker(config *Config) (*Broker, error) {
-	if config == nil {
-		config = DefaultConfig
-	}
-
 	b := &Broker{
 		id:          GenUniqueId(),
 		config:      config,
 		wpool:       pool.New(config.Worker),
 		nodes:       make(map[string]interface{}),
+		queues:      make(map[string]int),
 		clusterPool: make(chan *Message),
 	}

@@ -91,14 +91,19 @@ func NewBroker(config *Config) (*Broker, error) {
 		}
 		b.tlsConfig = tlsconfig
 	}
-
-	b.auth = auth.NewAuth(b.config.Plugin.Auth)
-	b.bridgeMQ = bridge.NewBridgeMQ(b.config.Plugin.Bridge)
-
+	if b.config.Acl {
+		aclconfig, err := acl.AclConfigLoad(b.config.AclConf)
+		if err != nil {
+			log.Error("Load acl conf error", zap.Error(err))
+			return nil, err
+		}
+		b.AclConfig = aclconfig
+		b.StartAclWatcher()
+	}
 	return b, nil
 }

-func (b *Broker) SubmitWork(clientId string, msg *Message) {
+func (b *Broker) SubmitWork(msg *Message) {
 	if b.wpool == nil {
 		b.wpool = pool.New(b.config.Worker)
 	}
@@ -106,7 +111,7 @@ func (b *Broker) SubmitWork(clientId string, msg *Message) {
 	if msg.client.typ == CLUSTER {
 		b.clusterPool <- msg
 	} else {
-		b.wpool.Submit(clientId, func() {
+		b.wpool.Submit(func() {
 			ProcessMessage(msg)
 		})
 	}
@@ -119,10 +124,6 @@ func (b *Broker) Start() {
 		return
 	}

-	if b.config.HTTPPort != "" {
-		go InitHTTPMoniter(b)
-	}
-
 	//listen clinet over tcp
 	if b.config.Port != "" {
 		go b.StartClientListening(false)
@@ -149,6 +150,22 @@ func (b *Broker) Start() {
 		b.ConnectToDiscovery()
 	}

+	//system monitor
+	go StateMonitor()
+
+}
+
+func StateMonitor() {
+	v, _ := mem.VirtualMemory()
+	timeSticker := time.NewTicker(time.Second * 30)
+	for {
+		select {
+		case <-timeSticker.C:
+			if v.UsedPercent > 75 {
+				debug.FreeOSMemory()
+			}
+		}
+	}
 }

 func (b *Broker) StartWebsocketListening() {
@@ -170,6 +187,7 @@ func (b *Broker) StartWebsocketListening() {

 func (b *Broker) wsHandler(ws *websocket.Conn) {
 	// io.Copy(ws, ws)
+	atomic.AddUint64(&b.cid, 1)
 	ws.PayloadType = websocket.BinaryFrame
 	b.handleConnection(CLIENT, ws)
 }
@@ -209,11 +227,41 @@ func (b *Broker) StartClientListening(Tls bool) {
 			continue
 		}
 		tmpDelay = ACCEPT_MIN_SLEEP
+		atomic.AddUint64(&b.cid, 1)
 		go b.handleConnection(CLIENT, conn)

 	}
 }

+func (b *Broker) Handshake(conn net.Conn) bool {
+
+	nc := tls.Server(conn, b.tlsConfig)
+	time.AfterFunc(DEFAULT_TLS_TIMEOUT, func() { TlsTimeout(nc) })
+	nc.SetReadDeadline(time.Now().Add(DEFAULT_TLS_TIMEOUT))
+
+	// Force handshake
+	if err := nc.Handshake(); err != nil {
+		log.Error("TLS handshake error, ", zap.Error(err))
+		return false
+	}
+	nc.SetReadDeadline(time.Time{})
+	return true
+
+}
+
+func TlsTimeout(conn *tls.Conn) {
+	nc := conn
+	// Check if already closed
+	if nc == nil {
+		return
+	}
+	cs := nc.ConnectionState()
+	if !cs.HandshakeComplete {
+		log.Error("TLS handshake timeout")
+		nc.Close()
+	}
+}
+
 func (b *Broker) StartClusterListening() {
 	var hp string = b.config.Cluster.Host + ":" + b.config.Cluster.Port
 	log.Info("Start Listening cluster on ", zap.String("hp", hp))
@@ -263,32 +311,9 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		log.Error("received msg that was not Connect")
 		return
 	}
-
-	log.Info("read connect from ", zap.String("clientID", msg.ClientIdentifier))
-
 	connack := packets.NewControlPacket(packets.Connack).(*packets.ConnackPacket)
+	connack.ReturnCode = packets.Accepted
 	connack.SessionPresent = msg.CleanSession
-	connack.ReturnCode = msg.Validate()
-
-	if connack.ReturnCode != packets.Accepted {
-		err = connack.Write(conn)
-		if err != nil {
-			log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
-			return
-		}
-		return
-	}
-
-	if typ == CLIENT && !b.CheckConnectAuth(string(msg.ClientIdentifier), string(msg.Username), string(msg.Password)) {
-		connack.ReturnCode = packets.ErrRefusedNotAuthorised
-		err = connack.Write(conn)
-		if err != nil {
-			log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
-			return
-		}
-		return
-	}
-
 	err = connack.Write(conn)
 	if err != nil {
 		log.Error("send connack error, ", zap.Error(err), zap.String("clientID", msg.ClientIdentifier))
@@ -346,14 +371,6 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		b.clients.Store(cid, c)

 		b.OnlineOfflineNotification(cid, true)
-		{
-			b.Publish(&bridge.Elements{
-				ClientID:  string(msg.ClientIdentifier),
-				Username:  string(msg.Username),
-				Action:    bridge.Connect,
-				Timestamp: time.Now().Unix(),
-			})
-		}
 	case ROUTER:
 		old, exist = b.routes.Load(cid)
 		if exist {
@@ -366,6 +383,8 @@ func (b *Broker) handleConnection(typ int, conn net.Conn) {
 		b.routes.Store(cid, c)
 	}

+	// mpool := b.messagePool[fnv1a.HashString64(cid)%MessagePoolNum]
+
 	c.readLoop()
 }

--- a/broker/client.go
+++ b/broker/client.go
@@ -1,21 +1,19 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
 	"context"
 	"errors"
-	"math/rand"
 	"net"
 	"reflect"
-	"regexp"
 	"strings"
 	"sync"
 	"time"

-	"github.com/fhmq/hmq/broker/lib/sessions"
-	"github.com/fhmq/hmq/broker/lib/topics"
-	"github.com/fhmq/hmq/plugins/bridge"
-
 	"github.com/eclipse/paho.mqtt.golang/packets"
+	"github.com/fhmq/hmq/lib/sessions"
+	"github.com/fhmq/hmq/lib/topics"
 	"go.uber.org/zap"
 )

@@ -30,45 +28,34 @@ const (
 	REMOTE  = 2
 	CLUSTER = 3
 )
-
-const (
-	_GroupTopicRegexp = `^\$share/([0-9a-zA-Z_-]+)/(.*)$`
-)
-
 const (
 	Connected    = 1
 	Disconnected = 2
 )

-var (
-	groupCompile = regexp.MustCompile(_GroupTopicRegexp)
-)
-
 type client struct {
-	typ         int
-	mu          sync.Mutex
-	broker      *Broker
-	conn        net.Conn
-	info        info
-	route       route
-	status      int
-	ctx         context.Context
-	cancelFunc  context.CancelFunc
-	session     *sessions.Session
-	subMap      map[string]*subscription
-	topicsMgr   *topics.Manager
-	subs        []interface{}
-	qoss        []byte
-	rmsgs       []*packets.PublishPacket
-	routeSubMap map[string]uint64
+	typ        int
+	mu         sync.Mutex
+	broker     *Broker
+	conn       net.Conn
+	info       info
+	route      route
+	status     int
+	ctx        context.Context
+	cancelFunc context.CancelFunc
+	session    *sessions.Session
+	subMap     map[string]*subscription
+	topicsMgr  *topics.Manager
+	subs       []interface{}
+	qoss       []byte
+	rmsgs      []*packets.PublishPacket
 }

 type subscription struct {
-	client    *client
-	topic     string
-	qos       byte
-	share     bool
-	groupName string
+	client *client
+	topic  string
+	qos    byte
+	queue  bool
 }

 type info struct {
@@ -88,13 +75,12 @@ type route struct {

 var (
 	DisconnectdPacket = packets.NewControlPacket(packets.Disconnect).(*packets.DisconnectPacket)
-	r                 = rand.New(rand.NewSource(time.Now().UnixNano()))
 )

 func (c *client) init() {
 	c.status = Connected
-	c.info.localIP, _, _ = net.SplitHostPort(c.conn.LocalAddr().String())
-	c.info.remoteIP, _, _ = net.SplitHostPort(c.conn.RemoteAddr().String())
+	c.info.localIP = strings.Split(c.conn.LocalAddr().String(), ":")[0]
+	c.info.remoteIP = strings.Split(c.conn.RemoteAddr().String(), ":")[0]
 	c.ctx, c.cancelFunc = context.WithCancel(context.Background())
 	c.subMap = make(map[string]*subscription)
 	c.topicsMgr = c.broker.topicsMgr
@@ -118,22 +104,22 @@ func (c *client) readLoop() {
 			//add read timeout
 			if err := nc.SetReadDeadline(time.Now().Add(timeOut)); err != nil {
 				log.Error("set read timeout error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-				msg := &Message{
-					client: c,
-					packet: DisconnectdPacket,
-				}
-				b.SubmitWork(c.info.clientID, msg)
 				return
 			}

 			packet, err := packets.ReadPacket(nc)
 			if err != nil {
 				log.Error("read packet error: ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-				msg := &Message{
-					client: c,
-					packet: DisconnectdPacket,
+				msg := &Message{client: c, packet: DisconnectdPacket}
+				b.SubmitWork(msg)
+
+				// remove subscriptions related to that client
+				for topic, sub := range c.subMap {
+					t := []byte(topic)
+					c.topicsMgr.Unsubscribe(t, sub)
+					c.session.RemoveTopic(topic)
+					delete(c.subMap, topic)
 				}
-				b.SubmitWork(c.info.clientID, msg)
 				return
 			}

@@ -141,7 +127,7 @@ func (c *client) readLoop() {
 				client: c,
 				packet: packet,
 			}
-			b.SubmitWork(c.info.clientID, msg)
+			b.SubmitWork(msg)
 		}
 	}

@@ -153,11 +139,7 @@ func ProcessMessage(msg *Message) {
 	if ca == nil {
 		return
 	}
-
-	if c.typ == CLIENT {
-		log.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID))
-	}
-
+	log.Debug("Recv message:", zap.String("message type", reflect.TypeOf(msg.packet).String()[9:]), zap.String("ClientID", c.info.clientID))
 	switch ca.(type) {
 	case *packets.ConnackPacket:
 	case *packets.ConnectPacket:
@@ -187,74 +169,21 @@ func ProcessMessage(msg *Message) {
 }

 func (c *client) ProcessPublish(packet *packets.PublishPacket) {
-	switch c.typ {
-	case CLIENT:
-		c.processClientPublish(packet)
-	case ROUTER:
-		c.processRouterPublish(packet)
-	case CLUSTER:
-		c.processRemotePublish(packet)
-	}
-
-}
-
-func (c *client) processRemotePublish(packet *packets.PublishPacket) {
 	if c.status == Disconnected {
 		return
 	}

 	topic := packet.TopicName
-	if topic == BrokerInfoTopic {
+	if topic == BrokerInfoTopic && c.typ == CLUSTER {
 		c.ProcessInfo(packet)
 		return
 	}

-}
-
-func (c *client) processRouterPublish(packet *packets.PublishPacket) {
-	if c.status == Disconnected {
-		return
-	}
-
-	switch packet.Qos {
-	case QosAtMostOnce:
-		c.ProcessPublishMessage(packet)
-	case QosAtLeastOnce:
-		puback := packets.NewControlPacket(packets.Puback).(*packets.PubackPacket)
-		puback.MessageID = packet.MessageID
-		if err := c.WriterPacket(puback); err != nil {
-			log.Error("send puback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-			return
-		}
-		c.ProcessPublishMessage(packet)
-	case QosExactlyOnce:
-		return
-	default:
-		log.Error("publish with unknown qos", zap.String("ClientID", c.info.clientID))
-		return
-	}
-
-}
-
-func (c *client) processClientPublish(packet *packets.PublishPacket) {
-
-	topic := packet.TopicName
-
-	if !c.broker.CheckTopicAuth(PUB, c.info.clientID, c.info.username, c.info.remoteIP, topic) {
+	if !c.CheckTopicAuth(PUB, topic) {
 		log.Error("Pub Topics Auth failed, ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
 		return
 	}

-	//publish kafka
-	c.broker.Publish(&bridge.Elements{
-		ClientID:  c.info.clientID,
-		Username:  c.info.username,
-		Action:    bridge.Publish,
-		Timestamp: time.Now().Unix(),
-		Payload:   string(packet.Payload),
-		Topic:     topic,
-	})
-
 	switch packet.Qos {
 	case QosAtMostOnce:
 		c.ProcessPublishMessage(packet)
@@ -276,6 +205,9 @@ func (c *client) processClientPublish(packet *packets.PublishPacket) {
 }

 func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
+	if c.status == Disconnected {
+		return
+	}

 	b := c.broker
 	if b == nil {
@@ -289,19 +221,20 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 		}
 	}

+	c.mu.Lock()
 	err := c.topicsMgr.Subscribers([]byte(packet.TopicName), packet.Qos, &c.subs, &c.qoss)
+	c.mu.Unlock()
 	if err != nil {
 		log.Error("Error retrieving subscribers list: ", zap.String("ClientID", c.info.clientID))
 		return
 	}

-	// fmt.Println("psubs num: ", len(c.subs))
+	// log.Info("psubs num: ", len(r.psubs))
 	if len(c.subs) == 0 {
 		return
 	}

-	var qsub []int
-	for i, sub := range c.subs {
+	for _, sub := range c.subs {
 		s, ok := sub.(*subscription)
 		if ok {
 			if s.client.typ == ROUTER {
@@ -309,34 +242,17 @@ func (c *client) ProcessPublishMessage(packet *packets.PublishPacket) {
 					continue
 				}
 			}
-			if s.share {
-				qsub = append(qsub, i)
-			} else {
-				publish(s, packet)
+			err := s.client.WriterPacket(packet)
+			if err != nil {
+				log.Error("process message for psub error,  ", zap.Error(err), zap.String("ClientID", c.info.clientID))
 			}
-
 		}

 	}

-	if len(qsub) > 0 {
-		idx := r.Intn(len(qsub))
-		sub := c.subs[qsub[idx]].(*subscription)
-		publish(sub, packet)
-	}
-
 }

 func (c *client) ProcessSubscribe(packet *packets.SubscribePacket) {
-	switch c.typ {
-	case CLIENT:
-		c.processClientSubscribe(packet)
-	case ROUTER:
-		c.processRouterSubscribe(packet)
-	}
-}
-
-func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
 	if c.status == Disconnected {
 		return
 	}
@@ -355,56 +271,25 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
 	for i, topic := range topics {
 		t := topic
 		//check topic auth for client
-		if !b.CheckTopicAuth(SUB, c.info.clientID, c.info.username, c.info.remoteIP, topic) {
+		if !c.CheckTopicAuth(SUB, topic) {
 			log.Error("Sub topic Auth failed: ", zap.String("topic", topic), zap.String("ClientID", c.info.clientID))
 			retcodes = append(retcodes, QosFailure)
 			continue
 		}

-		b.Publish(&bridge.Elements{
-			ClientID:  c.info.clientID,
-			Username:  c.info.username,
-			Action:    bridge.Subscribe,
-			Timestamp: time.Now().Unix(),
-			Topic:     topic,
-		})
-
-		groupName := ""
-		share := false
-		if strings.HasPrefix(topic, "$share/") {
-			substr := groupCompile.FindStringSubmatch(topic)
-			if len(substr) != 3 {
-				retcodes = append(retcodes, QosFailure)
-				continue
-			}
-			share = true
-			groupName = substr[1]
-			topic = substr[2]
-		}
-
-		if oldSub, exist := c.subMap[t]; exist {
-			c.topicsMgr.Unsubscribe([]byte(oldSub.topic), oldSub)
-			delete(c.subMap, t)
-		}
-
 		sub := &subscription{
-			topic:     topic,
-			qos:       qoss[i],
-			client:    c,
-			share:     share,
-			groupName: groupName,
+			topic:  t,
+			qos:    qoss[i],
+			client: c,
 		}

 		rqos, err := c.topicsMgr.Subscribe([]byte(topic), qoss[i], sub)
 		if err != nil {
-			log.Error("subscribe error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-			retcodes = append(retcodes, QosFailure)
-			continue
+			return
 		}

-		c.subMap[t] = sub
-
-		c.session.AddTopic(t, qoss[i])
+		c.subMap[topic] = sub
+		c.session.AddTopic(topic, qoss[i])
 		retcodes = append(retcodes, rqos)
 		c.topicsMgr.Retained([]byte(topic), &c.rmsgs)

@@ -418,7 +303,9 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
 		return
 	}
 	//broadcast subscribe message
-	go b.BroadcastSubOrUnsubMessage(packet)
+	if c.typ == CLIENT {
+		go b.BroadcastSubOrUnsubMessage(packet)
+	}

 	//process retain message
 	for _, rm := range c.rmsgs {
@@ -430,76 +317,7 @@ func (c *client) processClientSubscribe(packet *packets.SubscribePacket) {
 	}
 }

-func (c *client) processRouterSubscribe(packet *packets.SubscribePacket) {
-	if c.status == Disconnected {
-		return
-	}
-
-	b := c.broker
-	if b == nil {
-		return
-	}
-	topics := packet.Topics
-	qoss := packet.Qoss
-
-	suback := packets.NewControlPacket(packets.Suback).(*packets.SubackPacket)
-	suback.MessageID = packet.MessageID
-	var retcodes []byte
-
-	for i, topic := range topics {
-		t := topic
-		groupName := ""
-		share := false
-		if strings.HasPrefix(topic, "$share/") {
-			substr := groupCompile.FindStringSubmatch(topic)
-			if len(substr) != 3 {
-				retcodes = append(retcodes, QosFailure)
-				continue
-			}
-			share = true
-			groupName = substr[1]
-			topic = substr[2]
-		}
-
-		sub := &subscription{
-			topic:     topic,
-			qos:       qoss[i],
-			client:    c,
-			share:     share,
-			groupName: groupName,
-		}
-
-		rqos, err := c.topicsMgr.Subscribe([]byte(topic), qoss[i], sub)
-		if err != nil {
-			log.Error("subscribe error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-			retcodes = append(retcodes, QosFailure)
-			continue
-		}
-
-		c.subMap[t] = sub
-		addSubMap(c.routeSubMap, topic)
-		retcodes = append(retcodes, rqos)
-	}
-
-	suback.ReturnCodes = retcodes
-
-	err := c.WriterPacket(suback)
-	if err != nil {
-		log.Error("send suback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-		return
-	}
-}
-
 func (c *client) ProcessUnSubscribe(packet *packets.UnsubscribePacket) {
-	switch c.typ {
-	case CLIENT:
-		c.processClientUnSubscribe(packet)
-	case ROUTER:
-		c.processRouterUnSubscribe(packet)
-	}
-}
-
-func (c *client) processRouterUnSubscribe(packet *packets.UnsubscribePacket) {
 	if c.status == Disconnected {
 		return
 	}
@@ -510,60 +328,13 @@ func (c *client) processRouterUnSubscribe(packet *packets.UnsubscribePacket) {
 	topics := packet.Topics

 	for _, topic := range topics {
+		t := []byte(topic)
 		sub, exist := c.subMap[topic]
 		if exist {
-			retainNum := delSubMap(c.routeSubMap, topic)
-			if retainNum > 0 {
-				continue
-			}
-
-			c.topicsMgr.Unsubscribe([]byte(sub.topic), sub)
-			delete(c.subMap, topic)
-		}
-
-	}
-
-	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
-	unsuback.MessageID = packet.MessageID
-
-	err := c.WriterPacket(unsuback)
-	if err != nil {
-		log.Error("send unsuback error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-		return
-	}
-}
-
-func (c *client) processClientUnSubscribe(packet *packets.UnsubscribePacket) {
-	if c.status == Disconnected {
-		return
-	}
-	b := c.broker
-	if b == nil {
-		return
-	}
-	topics := packet.Topics
-
-	for _, topic := range topics {
-		{
-			//publish kafka
-
-			b.Publish(&bridge.Elements{
-				ClientID:  c.info.clientID,
-				Username:  c.info.username,
-				Action:    bridge.Unsubscribe,
-				Timestamp: time.Now().Unix(),
-				Topic:     topic,
-			})
-
-		}
-
-		sub, exist := c.subMap[topic]
-		if exist {
-			c.topicsMgr.Unsubscribe([]byte(sub.topic), sub)
+			c.topicsMgr.Unsubscribe(t, sub)
 			c.session.RemoveTopic(topic)
 			delete(c.subMap, topic)
 		}
-
 	}

 	unsuback := packets.NewControlPacket(packets.Unsuback).(*packets.UnsubackPacket)
@@ -575,7 +346,9 @@ func (c *client) processClientUnSubscribe(packet *packets.UnsubscribePacket) {
 		return
 	}
 	// //process ubsubscribe message
-	b.BroadcastSubOrUnsubMessage(packet)
+	if c.typ == CLIENT {
+		b.BroadcastSubOrUnsubMessage(packet)
+	}
 }

 func (c *client) ProcessPing() {
@@ -599,32 +372,18 @@ func (c *client) Close() {

 	c.status = Disconnected
 	//wait for message complete
-	// time.Sleep(1 * time.Second)
+	time.Sleep(1 * time.Second)
 	// c.status = Disconnected

-	b := c.broker
-	b.Publish(&bridge.Elements{
-		ClientID:  c.info.clientID,
-		Username:  c.info.username,
-		Action:    bridge.Disconnect,
-		Timestamp: time.Now().Unix(),
-	})
-
 	if c.conn != nil {
 		c.conn.Close()
 		c.conn = nil
 	}

+	b := c.broker
 	subs := c.subMap
-
 	if b != nil {
 		b.removeClient(c)
-		for _, sub := range subs {
-			err := b.topicsMgr.Unsubscribe([]byte(sub.topic), sub)
-			if err != nil {
-				log.Error("unsubscribe error, ", zap.Error(err), zap.String("ClientID", c.info.clientID))
-			}
-		}

 		if c.typ == CLIENT {
 			b.BroadcastUnSubscribe(subs)
--- a/broker/comm.go
+++ b/broker/comm.go
@@ -1,15 +1,15 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
-	"encoding/json"
+	"crypto/md5"
+	"crypto/rand"
+	"encoding/base64"
+	"encoding/hex"
+	"io"
 	"reflect"
 	"time"
-
-	"github.com/tidwall/gjson"
-	"go.uber.org/zap"
-
-	"github.com/eclipse/paho.mqtt.golang/packets"
-	uuid "github.com/satori/go.uuid"
 )

 const (
@@ -91,65 +91,13 @@ func equal(k1, k2 interface{}) bool {
 	return false
 }

-func addSubMap(m map[string]uint64, topic string) {
-	subNum, exist := m[topic]
-	if exist {
-		m[topic] = subNum + 1
-	} else {
-		m[topic] = 1
-	}
-}
-
-func delSubMap(m map[string]uint64, topic string) uint64 {
-	subNum, exist := m[topic]
-	if exist {
-		if subNum > 1 {
-			m[topic] = subNum - 1
-			return subNum - 1
-		}
-	} else {
-		m[topic] = 0
-	}
-	return 0
-}
-
 func GenUniqueId() string {
-	return uuid.NewV4().String()
-}
-
-func wrapPublishPacket(packet *packets.PublishPacket) *packets.PublishPacket {
-	p := packet.Copy()
-	wrapPayload := map[string]interface{}{
-		"message_id": GenUniqueId(),
-		"payload":    string(p.Payload),
-	}
-	b, _ := json.Marshal(wrapPayload)
-	p.Payload = b
-	return p
-}
-
-func unWrapPublishPacket(packet *packets.PublishPacket) *packets.PublishPacket {
-	p := packet.Copy()
-	if gjson.GetBytes(p.Payload, "payload").Exists() {
-		p.Payload = []byte(gjson.GetBytes(p.Payload, "payload").String())
-	}
-	return p
-}
-
-func publish(sub *subscription, packet *packets.PublishPacket) {
-	// var p *packets.PublishPacket
-	// if sub.client.info.username != "root" {
-	// 	p = unWrapPublishPacket(packet)
-	// } else {
-	// 	p = wrapPublishPacket(packet)
-	// }
-	// err := sub.client.WriterPacket(p)
-	// if err != nil {
-	// 	log.Error("process message for psub error,  ", zap.Error(err))
-	// }
-
-	err := sub.client.WriterPacket(packet)
-	if err != nil {
-		log.Error("process message for psub error,  ", zap.Error(err))
+	b := make([]byte, 48)
+	if _, err := io.ReadFull(rand.Reader, b); err != nil {
+		return ""
 	}
+	h := md5.New()
+	h.Write([]byte(base64.URLEncoding.EncodeToString(b)))
+	return hex.EncodeToString(h.Sum(nil))
+	// return GetMd5String()
 }
--- a/broker/config.go
+++ b/broker/config.go
@@ -1,3 +1,5 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
@@ -15,25 +17,20 @@ import (
 )

 type Config struct {
-	Worker   int       `json:"workerNum"`
-	HTTPPort string    `json:"httpPort"`
-	Host     string    `json:"host"`
-	Port     string    `json:"port"`
-	Cluster  RouteInfo `json:"cluster"`
-	Router   string    `json:"router"`
-	TlsHost  string    `json:"tlsHost"`
-	TlsPort  string    `json:"tlsPort"`
-	WsPath   string    `json:"wsPath"`
-	WsPort   string    `json:"wsPort"`
-	WsTLS    bool      `json:"wsTLS"`
-	TlsInfo  TLSInfo   `json:"tlsInfo"`
-	Debug    bool      `json:"debug"`
-	Plugin   Plugins   `json:"plugins"`
-}
-
-type Plugins struct {
-	Auth   string
-	Bridge string
+	Worker  int       `json:"workerNum"`
+	Host    string    `json:"host"`
+	Port    string    `json:"port"`
+	Cluster RouteInfo `json:"cluster"`
+	Router  string    `json:"router"`
+	TlsHost string    `json:"tlsHost"`
+	TlsPort string    `json:"tlsPort"`
+	WsPath  string    `json:"wsPath"`
+	WsPort  string    `json:"wsPort"`
+	WsTLS   bool      `json:"wsTLS"`
+	TlsInfo TLSInfo   `json:"tlsInfo"`
+	Acl     bool      `json:"acl"`
+	AclConf string    `json:"aclConf"`
+	Debug   bool      `json:"-"`
 }

 type RouteInfo struct {
@@ -52,10 +49,11 @@ var DefaultConfig *Config = &Config{
 	Worker: 4096,
 	Host:   "0.0.0.0",
 	Port:   "1883",
+	Acl:    false,
 }

 var (
-	log = logger.Prod().Named("broker")
+	log *zap.Logger
 )

 func showHelp() {
@@ -76,8 +74,6 @@ func ConfigureConfig(args []string) (*Config, error) {
 	fs.BoolVar(&help, "help", false, "Show this message.")
 	fs.IntVar(&config.Worker, "w", 1024, "worker num to process message, perfer (client num)/10.")
 	fs.IntVar(&config.Worker, "worker", 1024, "worker num to process message, perfer (client num)/10.")
-	fs.StringVar(&config.HTTPPort, "httpport", "8080", "Port to listen on.")
-	fs.StringVar(&config.HTTPPort, "hp", "8080", "Port to listen on.")
 	fs.StringVar(&config.Port, "port", "1883", "Port to listen on.")
 	fs.StringVar(&config.Port, "p", "1883", "Port to listen on.")
 	fs.StringVar(&config.Host, "host", "0.0.0.0", "Network host to listen on")
@@ -112,6 +108,9 @@ func ConfigureConfig(args []string) (*Config, error) {
 		}
 	})

+	logger.InitLogger(config.Debug)
+	log = logger.Get().Named("Broker")
+
 	if configFile != "" {
 		tmpConfig, e := LoadConfig(configFile)
 		if e != nil {
@@ -121,10 +120,6 @@ func ConfigureConfig(args []string) (*Config, error) {
 		}
 	}

-	if config.Debug {
-		log = logger.Debug().Named("broker")
-	}
-
 	if err := config.check(); err != nil {
 		return nil, err
 	}
@@ -137,7 +132,7 @@ func LoadConfig(filename string) (*Config, error) {

 	content, err := ioutil.ReadFile(filename)
 	if err != nil {
-		// log.Error("Read config file error: ", zap.Error(err))
+		log.Error("Read config file error: ", zap.Error(err))
 		return nil, err
 	}
 	// log.Info(string(content))
@@ -145,7 +140,7 @@ func LoadConfig(filename string) (*Config, error) {
 	var config Config
 	err = json.Unmarshal(content, &config)
 	if err != nil {
-		// log.Error("Unmarshal config file error: ", zap.Error(err))
+		log.Error("Unmarshal config file error: ", zap.Error(err))
 		return nil, err
 	}

--- a/broker/http.go
+++ b/broker/http.go
@@ -1,26 +0,0 @@
-package broker
-
-import (
-	"github.com/gin-gonic/gin"
-)
-
-func InitHTTPMoniter(b *Broker) {
-	gin.SetMode(gin.ReleaseMode)
-	router := gin.Default()
-	router.DELETE("api/v1/connections/:clientid", func(c *gin.Context) {
-		clientid := c.Param("clientid")
-		cli, ok := b.clients.Load(clientid)
-		if ok {
-			conn, succss := cli.(*client)
-			if succss {
-				conn.Close()
-			}
-		}
-		resp := map[string]int{
-			"code": 0,
-		}
-		c.JSON(200, &resp)
-	})
-
-	router.Run(":" + b.config.HTTPPort)
-}
--- a/broker/info.go
+++ b/broker/info.go
@@ -1,3 +1,5 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package broker

 import (
--- a/plugins/auth/authfile/acl.conf
+++ b/plugins/auth/authfile/acl.conf
@@ -1,4 +1,4 @@
-## sub 1 ,  pub 2,  pubsub 3
+## pub 1 ,  sub 2,  pubsub 3
 ## %c is clientid , %s is username
 ##auth     type        value       pub/sub      topic
 allow      ip          127.0.0.1      2         $SYS/#       
@@ -9,4 +9,4 @@ allow      clientid    *              1         toCloud/%c
 allow      username    *              1         toCloud/%u   
 allow      clientid    *              2         toDevice/%c  
 allow      username    *              2         toDevice/%u  
-deny       clientid    *              3         #     
+deny       clientid    *              3         #            
--- a/conf/hmq.config
+++ b/conf/hmq.config
@@ -2,12 +2,10 @@
 	"workerNum": 4096,
 	"port": "1883",
 	"host": "0.0.0.0",
-	"debug": true,
 	"cluster": {
 		"host": "0.0.0.0",
 		"port": "1993"
 	},
-	"httpPort": "8080",
 	"router": "127.0.0.1:9888",
 	"tlsPort": "8883",
 	"tlsHost": "0.0.0.0",
@@ -20,8 +18,6 @@
 		"certFile": "ssl/server/cert.pem",
 		"keyFile": "ssl/server/key.pem"
 	},
-	"plugins": {
-		"auth": "authhttp",
-		"bridge": "kafka"
-	}
+	"acl": false,
+	"aclConf": "conf/acl.conf"
 }
--- a/deploy/config.yaml
+++ b/deploy/config.yaml
@@ -1,37 +0,0 @@
-apiVersion: v1
-kind: ConfigMap
-metadata:
-  name: mqtt-broker
-data:
-  hmq.config: |
-    {
-      "workerNum": 4096,
-      "port": "1883",
-      "host": "0.0.0.0",
-      "plugins": ["authhttp","kafka"]
-    }
-
-  kafka.json: |
-    {
-        "addr": [
-            "127.0.0.1:9090"
-        ],
-        "onConnect": "onConnect",
-        "onPublish": "onPublish",
-        "onSubscribe": "onSubscribe",
-        "onDisconnect": "onDisconnect",
-        "onUnsubscribe": "onUnsubscribe",
-        "deliverMap": {
-          "#": "publish",
-          "/upload/+/#": "upload"
-        }
-    }
-  
-  authhttp.json: |
-    {
-        "auth": "http://127.0.0.1:9090/mqtt/auth",
-        "acl": "http://127.0.0.1:9090/mqtt/acl",
-        "super": "http://127.0.0.1:9090/mqtt/superuser"
-    }
-    
-
--- a/deploy/deploy.yaml
+++ b/deploy/deploy.yaml
@@ -1,44 +0,0 @@
-apiVersion: apps/v1
-kind: Deployment
-metadata:
-  name: mqtt-broker
-spec:
-  selector:
-    matchLabels:
-      app: mqtt-broker
-  replicas: 1
-  template:
-    metadata:
-      labels:
-        app: mqtt-broker
-    spec:
-      containers:
-        - name: mqtt-broker
-          image: uhub.service.ucloud.cn/uiot_core_hub/hmq:v0.1.0
-          ports:
-            - containerPort: 1883
-            - containerPort: 8080
-          volumeMounts:
-            - name: mqtt-broker
-              mountPath: /conf
-              subPath: hmq.config
-            - name: mqtt-broker
-              mountPath: /plugins/kafka/kafka.json
-              subPath: kafka.json
-            - name: mqtt-broker
-              mountPath: /plugins/authttp/http.json
-              subPath: kafka.json     
-      volumes:
-        - name: mqtt-broker
-          configMap:
-            name: mqtt-broker
-            items:
-              - key: hmq.config
-                path: hmq.config
-            items:
-              - key: http.json
-                path: http.json
-            items:
-              - key: kafka.json
-                path: kafka.json
-
--- a/deploy/svc.yaml
+++ b/deploy/svc.yaml
@@ -1,13 +0,0 @@
-kind: Service
-apiVersion: v1
-metadata:
-  name: mqtt-broker
-spec:
-  selector:
-    app: mqtt-broker
-  ports:
-    - protocol: TCP
-      port: 1883
-      targetPort: 1883
-  type: ClusterIP
-  sessionAffinity: ClientIP
--- a/go.mod
+++ b/go.mod
@@ -1,27 +0,0 @@
-module github.com/fhmq/hmq
-
-go 1.12
-
-require (
-	github.com/Shopify/sarama v1.26.1
-	github.com/bitly/go-simplejson v0.5.0
-	github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 // indirect
-	github.com/eclipse/paho.mqtt.golang v1.2.0
-	github.com/gin-gonic/gin v1.4.0
-	github.com/golang/protobuf v1.3.2 // indirect
-	github.com/kr/pretty v0.1.0 // indirect
-	github.com/patrickmn/go-cache v2.1.0+incompatible
-	github.com/pkg/errors v0.8.1 // indirect
-	github.com/satori/go.uuid v1.2.0
-	github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e
-	github.com/stretchr/testify v1.3.0
-	github.com/tidwall/gjson v1.3.0
-	go.uber.org/atomic v1.4.0 // indirect
-	go.uber.org/multierr v1.1.0 // indirect
-	go.uber.org/zap v1.10.0
-	golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 // indirect
-	golang.org/x/net v0.0.0-20190724013045-ca1201d0de80
-	golang.org/x/sys v0.0.0-20190730183949-1393eb018365 // indirect
-	gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 // indirect
-	gopkg.in/jcmturner/goidentity.v3 v3.0.0 // indirect
-)
--- a/go.sum
+++ b/go.sum
@@ -1,119 +0,0 @@
-github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798 h1:2T/jmrHeTezcCM58lvEQXs0UpQJCo5SoGAcg+mbSTIg=
-github.com/DataDog/zstd v1.3.6-0.20190409195224-796139022798/go.mod h1:1jcaCB/ufaK+sKp1NBhlGmpz41jOoPQ35bpF36t7BBo=
-github.com/Shopify/sarama v1.23.0 h1:slvlbm7bxyp7sKQbUwha5BQdZTqurhRoI+zbKorVigQ=
-github.com/Shopify/sarama v1.23.0/go.mod h1:XLH1GYJnLVE0XCr6KdJGVJRTwY30moWNJ4sERjXX6fs=
-github.com/Shopify/toxiproxy v2.1.4+incompatible h1:TKdv8HiTLgE5wdJuEML90aBgNWsokNbMijUGhmcoBJc=
-github.com/Shopify/toxiproxy v2.1.4+incompatible/go.mod h1:OXgGpZ6Cli1/URJOF1DMxUHB2q5Ap20/P/eIdh4G0pI=
-github.com/bitly/go-simplejson v0.5.0 h1:6IH+V8/tVMab511d5bn4M7EwGXZf9Hj6i2xSwkNEM+Y=
-github.com/bitly/go-simplejson v0.5.0/go.mod h1:cXHtHw4XUPsvGaxgjIAn8PhEWG9NfngEKAMDJEczWVA=
-github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869 h1:DDGfHa7BWjL4YnC6+E63dPcxHo2sUxDIu8g3QgEJdRY=
-github.com/bmizerany/assert v0.0.0-20160611221934-b7ed37b82869/go.mod h1:Ekp36dRnpXw/yCqJaO+ZrUyxD+3VXMFFr56k5XYrpB4=
-github.com/davecgh/go-spew v1.1.0 h1:ZDRjVQ15GmhC3fiQ8ni8+OwkZQO4DARzQgrnXU1Liz8=
-github.com/davecgh/go-spew v1.1.0/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/davecgh/go-spew v1.1.1 h1:vj9j/u1bqnvCEfJOwUhtlOARqs3+rkHYY13jYWTU97c=
-github.com/davecgh/go-spew v1.1.1/go.mod h1:J7Y8YcW2NihsgmVo/mv3lAwl/skON4iLHjSsI+c5H38=
-github.com/eapache/go-resiliency v1.1.0 h1:1NtRmCAqadE2FN4ZcN6g90TP3uk8cg9rn9eNK2197aU=
-github.com/eapache/go-resiliency v1.1.0/go.mod h1:kFI+JgMyC7bLPUVY133qvEBtVayf5mFgVsvEsIPBvNs=
-github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21 h1:YEetp8/yCZMuEPMUDHG0CW/brkkEp8mzqk2+ODEitlw=
-github.com/eapache/go-xerial-snappy v0.0.0-20180814174437-776d5712da21/go.mod h1:+020luEh2TKB4/GOp8oxxtq0Daoen/Cii55CzbTV6DU=
-github.com/eapache/queue v1.1.0 h1:YOEu7KNc61ntiQlcEeUIoDTJ2o8mQznoNvUhiigpIqc=
-github.com/eapache/queue v1.1.0/go.mod h1:6eCeP0CKFpHLu8blIFXhExK/dRa7WDZfr6jVFPTqq+I=
-github.com/eclipse/paho.mqtt.golang v1.2.0 h1:1F8mhG9+aO5/xpdtFkW4SxOJB67ukuDC3t2y2qayIX0=
-github.com/eclipse/paho.mqtt.golang v1.2.0/go.mod h1:H9keYFcgq3Qr5OUJm/JZI/i6U7joQ8SYLhZwfeOo6Ts=
-github.com/gin-contrib/sse v0.0.0-20190301062529-5545eab6dad3 h1:t8FVkw33L+wilf2QiWkw0UV77qRpcH/JHPKGpKa2E8g=
-github.com/gin-contrib/sse v0.0.0-20190301062529-5545eab6dad3/go.mod h1:VJ0WA2NBN22VlZ2dKZQPAPnyWw5XTlK1KymzLKsr59s=
-github.com/gin-gonic/gin v1.4.0 h1:3tMoCCfM7ppqsR0ptz/wi1impNpT7/9wQtMZ8lr1mCQ=
-github.com/gin-gonic/gin v1.4.0/go.mod h1:OW2EZn3DO8Ln9oIKOvM++LBO+5UPHJJDH72/q/3rZdM=
-github.com/golang/protobuf v1.3.1 h1:YF8+flBXS5eO826T4nzqPrxfhQThhXl0YzfuUPu4SBg=
-github.com/golang/protobuf v1.3.1/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/protobuf v1.3.2 h1:6nsPYzhq5kReh6QImI3k5qWzO4PEbvbIW2cwSfR/6xs=
-github.com/golang/protobuf v1.3.2/go.mod h1:6lQm79b+lXiMfvg/cZm0SGofjICqVBUtrP5yJMmIC1U=
-github.com/golang/snappy v0.0.1 h1:Qgr9rKW7uDUkrbSmQeiDsGa8SjGyCOGtuasMWwvp2P4=
-github.com/golang/snappy v0.0.1/go.mod h1:/XxbfmMg8lxefKM7IXC3fBNl/7bRcc72aCRzEWrmP2Q=
-github.com/hashicorp/go-uuid v1.0.1 h1:fv1ep09latC32wFoVwnqcnKJGnMSdBanPczbHAYm1BE=
-github.com/hashicorp/go-uuid v1.0.1/go.mod h1:6SBZvOh/SIDV7/2o3Jml5SYk/TvGqwFJ/bN7x4byOro=
-github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03 h1:FUwcHNlEqkqLjLBdCp5PRlCFijNjvcYANOZXzCfXwCM=
-github.com/jcmturner/gofork v0.0.0-20190328161633-dc7c13fece03/go.mod h1:MK8+TM0La+2rjBD4jE12Kj1pCCxK7d2LK/UM3ncEo0o=
-github.com/json-iterator/go v1.1.6 h1:MrUvLMLTMxbqFJ9kzlvat/rYZqZnW3u4wkLzWTaFwKs=
-github.com/json-iterator/go v1.1.6/go.mod h1:+SdeFBvtyEkXs7REEP0seUULqWtbJapLOCVDaaPEHmU=
-github.com/kr/pretty v0.1.0 h1:L/CwN0zerZDmRFUapSPitk6f+Q3+0za1rQkzVuMiMFI=
-github.com/kr/pretty v0.1.0/go.mod h1:dAy3ld7l9f0ibDNOQOHHMYYIIbhfbHSm3C4ZsoJORNo=
-github.com/kr/pty v1.1.1/go.mod h1:pFQYn66WHrOpPYNljwOMqo10TkYh1fy3cYio2l3bCsQ=
-github.com/kr/text v0.1.0 h1:45sCR5RtlFHMR4UwH9sdQ5TC8v0qDQCHnXt+kaKSTVE=
-github.com/kr/text v0.1.0/go.mod h1:4Jbv+DJW3UT/LiOwJeYQe1efqtUx/iVham/4vfdArNI=
-github.com/mattn/go-isatty v0.0.7 h1:UvyT9uN+3r7yLEYSlJsbQGdsaB/a0DlgWP3pql6iwOc=
-github.com/mattn/go-isatty v0.0.7/go.mod h1:Iq45c/XA43vh69/j3iqttzPXn0bhXyGjM0Hdxcsrc5s=
-github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd h1:TRLaZ9cD/w8PVh93nsPXa1VrQ6jlwL5oN8l14QlcNfg=
-github.com/modern-go/concurrent v0.0.0-20180306012644-bacd9c7ef1dd/go.mod h1:6dJC0mAP4ikYIbvyc7fijjWJddQyLn8Ig3JB5CqoB9Q=
-github.com/modern-go/reflect2 v1.0.1 h1:9f412s+6RmYXLWZSEzVVgPGK7C2PphHj5RJrvfx9AWI=
-github.com/modern-go/reflect2 v1.0.1/go.mod h1:bx2lNnkwVCuqBIxFjflWJWanXIb3RllmbCylyMrvgv0=
-github.com/patrickmn/go-cache v2.1.0+incompatible h1:HRMgzkcYKYpi3C8ajMPV8OFXaaRUnok+kx1WdO15EQc=
-github.com/patrickmn/go-cache v2.1.0+incompatible/go.mod h1:3Qf8kWWT7OJRJbdiICTKqZju1ZixQ/KpMGzzAfe6+WQ=
-github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41 h1:GeinFsrjWz97fAxVUEd748aV0cYL+I6k44gFJTCVvpU=
-github.com/pierrec/lz4 v0.0.0-20190327172049-315a67e90e41/go.mod h1:3/3N9NVKO0jef7pBehbT1qWhCMrIgbYNnFAZCqQ5LRc=
-github.com/pkg/errors v0.8.1 h1:iURUrRGxPUNPdy5/HRSm+Yj6okJ6UtLINN0Q9M4+h3I=
-github.com/pkg/errors v0.8.1/go.mod h1:bwawxfHBFNV+L2hUp1rHADufV3IMtnDRdf1r5NINEl0=
-github.com/pkg/profile v1.2.1/go.mod h1:hJw3o1OdXxsrSjjVksARp5W95eeEaEfptyVZyv6JUPA=
-github.com/pmezard/go-difflib v1.0.0 h1:4DBwDE0NGyQoBHbLQYPwSUPoCMWR5BEzIk/f1lZbAQM=
-github.com/pmezard/go-difflib v1.0.0/go.mod h1:iKH77koFhYxTK1pcRnkKkqfTogsbg7gZNVY4sRDYZ/4=
-github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a h1:9ZKAASQSHhDYGoxY8uLVpewe1GDZ2vu2Tr/vTdVAkFQ=
-github.com/rcrowley/go-metrics v0.0.0-20181016184325-3113b8401b8a/go.mod h1:bCqnVzQkZxMG4s8nGwiZ5l3QUCyqpo9Y+/ZMZ9VjZe4=
-github.com/satori/go.uuid v1.2.0 h1:0uYX9dsZ2yD7q2RtLRtPSdGDWzjeM3TbMJP9utgA0ww=
-github.com/satori/go.uuid v1.2.0/go.mod h1:dA0hQrYB0VpLJoorglMZABFdXlWrHn1NEOzdhQKdks0=
-github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e h1:uO75wNGioszjmIzcY/tvdDYKRLVvzggtAmmJkn9j4GQ=
-github.com/segmentio/fasthash v0.0.0-20180216231524-a72b379d632e/go.mod h1:tm/wZFQ8e24NYaBGIlnO2WGCAi67re4HHuOm0sftE/M=
-github.com/stretchr/objx v0.1.0/go.mod h1:HFkY916IF+rwdDfMAkV7OtwuqBVzrE8GR6GFx+wExME=
-github.com/stretchr/testify v1.3.0 h1:TivCn/peBQ7UY8ooIcPgZFpTNSz0Q2U6UrFlUfqbe0Q=
-github.com/stretchr/testify v1.3.0/go.mod h1:M5WIy9Dh21IEIfnGCwXGc5bZfKNJtfHm1UVUgZn+9EI=
-github.com/tidwall/gjson v1.3.0 h1:kfpsw1W3trbg4Xm6doUtqSl9+LhLB6qJ9PkltVAQZYs=
-github.com/tidwall/gjson v1.3.0/go.mod h1:P256ACg0Mn+j1RXIDXoss50DeIABTYK1PULOJHhxOls=
-github.com/tidwall/match v1.0.1 h1:PnKP62LPNxHKTwvHHZZzdOAOCtsJTjo6dZLCwpKm5xc=
-github.com/tidwall/match v1.0.1/go.mod h1:LujAq0jyVjBy028G1WhWfIzbpQfMO8bBZ6Tyb0+pL9E=
-github.com/tidwall/pretty v1.0.0 h1:HsD+QiTn7sK6flMKIvNmpqz1qrpP3Ps6jOKIKMooyg4=
-github.com/tidwall/pretty v1.0.0/go.mod h1:XNkn88O1ChpSDQmQeStsy+sBenx6DDtFZJxhVysOjyk=
-github.com/ugorji/go v1.1.4 h1:j4s+tAvLfL3bZyefP2SEWmhBzmuIlH/eqNuPdFPgngw=
-github.com/ugorji/go v1.1.4/go.mod h1:uQMGLiO92mf5W77hV/PUCpI3pbzQx3CRekS0kk+RGrc=
-github.com/xdg/scram v0.0.0-20180814205039-7eeb5667e42c/go.mod h1:lB8K/P019DLNhemzwFU4jHLhdvlE6uDZjXFejJXr49I=
-github.com/xdg/stringprep v1.0.0/go.mod h1:Jhud4/sHMO4oL310DaZAKk9ZaJ08SJfe+sJh0HrGL1Y=
-go.uber.org/atomic v1.4.0 h1:cxzIVoETapQEqDhQu3QfnvXAV4AlzcvUCxkVUFw3+EU=
-go.uber.org/atomic v1.4.0/go.mod h1:gD2HeocX3+yG+ygLZcrzQJaqmWj9AIm7n08wl/qW/PE=
-go.uber.org/multierr v1.1.0 h1:HoEmRHQPVSqub6w2z2d2EOVs2fjyFRGyofhKuyDq0QI=
-go.uber.org/multierr v1.1.0/go.mod h1:wR5kodmAFQ0UK8QlbwjlSNy0Z68gJhDJUG5sjR94q/0=
-go.uber.org/zap v1.10.0 h1:ORx85nbTijNz8ljznvCMR1ZBIPKFn3jQrag10X2AsuM=
-go.uber.org/zap v1.10.0/go.mod h1:vwi/ZaCAaUcBkycHslxD9B2zi4UTXhF60s6SWpuDF0Q=
-golang.org/x/crypto v0.0.0-20190308221718-c2843e01d9a2/go.mod h1:djNgcEr1/C05ACkg1iLfiJU5Ep61QUkGW8qpdssI0+w=
-golang.org/x/crypto v0.0.0-20190404164418-38d8ce5564a5/go.mod h1:WFFai1msRO1wXaEeE5yQxYXgSfI8pQAWXbQop6sCtWE=
-golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4 h1:HuIa8hRrWRSrqYzx1qI49NNxhdi2PrY7gxVSq1JjLDc=
-golang.org/x/crypto v0.0.0-20190701094942-4def268fd1a4/go.mod h1:yigFU9vqHzYiE8UmvKecakEJjdnWj3jj499lnFckfCI=
-golang.org/x/net v0.0.0-20190404232315-eb5bcb51f2a3/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c h1:uOCk1iQW6Vc18bnC13MfzScl+wdKBmM9Y9kU7Z83/lw=
-golang.org/x/net v0.0.0-20190503192946-f4e77d36d62c/go.mod h1:t9HGtf8HONx5eT2rtn7q6eTqICYqUVnKs3thJo3Qplg=
-golang.org/x/net v0.0.0-20190724013045-ca1201d0de80 h1:Ao/3l156eZf2AW5wK8a7/smtodRU+gha3+BeqJ69lRk=
-golang.org/x/net v0.0.0-20190724013045-ca1201d0de80/go.mod h1:z5CRVTTTmAJ677TzLLGU+0bjPO0LkuOLi4/5GtJWs/s=
-golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a h1:1BGLXjeY4akVXGgbC9HugT3Jv3hCI0z56oJR5vAMgBU=
-golang.org/x/sys v0.0.0-20190215142949-d0b11bdaac8a/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190222072716-a9d3bda3a223/go.mod h1:STP8DvDyc/dI5b8T5hshtkjS+E42TnysNCUPdjciGhY=
-golang.org/x/sys v0.0.0-20190403152447-81d4e9dc473e/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190412213103-97732733099d/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/sys v0.0.0-20190730183949-1393eb018365 h1:SaXEMXhWzMJThc05vu6uh61Q245r4KaWMrsTedk0FDc=
-golang.org/x/sys v0.0.0-20190730183949-1393eb018365/go.mod h1:h1NjWce9XRLGQEsW7wpKNCjG9DtNlClVuFLEZdDNbEs=
-golang.org/x/text v0.3.0/go.mod h1:NqM8EUOU14njkJ3fqMW+pc6Ldnwhi/IjpwHt7yyuwOQ=
-gopkg.in/check.v1 v0.0.0-20161208181325-20d25e280405/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127 h1:qIbj1fsPNlZgppZ+VLlY7N33q108Sa+fhmuc+sWQYwY=
-gopkg.in/check.v1 v1.0.0-20180628173108-788fd7840127/go.mod h1:Co6ibVJAznAaIkqp8huTwlJQCZ016jof/cbN4VW5Yz0=
-gopkg.in/go-playground/assert.v1 v1.2.1 h1:xoYuJVE7KT85PYWrN730RguIQO0ePzVRfFMXadIrXTM=
-gopkg.in/go-playground/assert.v1 v1.2.1/go.mod h1:9RXL0bg/zibRAgZUYszZSwO/z8Y/a8bDuhia5mkpMnE=
-gopkg.in/go-playground/validator.v8 v8.18.2 h1:lFB4DoMU6B626w8ny76MV7VX6W2VHct2GVOI3xgiMrQ=
-gopkg.in/go-playground/validator.v8 v8.18.2/go.mod h1:RX2a/7Ha8BgOhfk7j780h4/u/RRjR0eouCJSH80/M2Y=
-gopkg.in/jcmturner/aescts.v1 v1.0.1 h1:cVVZBK2b1zY26haWB4vbBiZrfFQnfbTVrE3xZq6hrEw=
-gopkg.in/jcmturner/aescts.v1 v1.0.1/go.mod h1:nsR8qBOg+OucoIW+WMhB3GspUQXq9XorLnQb9XtvcOo=
-gopkg.in/jcmturner/dnsutils.v1 v1.0.1 h1:cIuC1OLRGZrld+16ZJvvZxVJeKPsvd5eUIvxfoN5hSM=
-gopkg.in/jcmturner/dnsutils.v1 v1.0.1/go.mod h1:m3v+5svpVOhtFAP/wSz+yzh4Mc0Fg7eRhxkJMWSIz9Q=
-gopkg.in/jcmturner/goidentity.v3 v3.0.0 h1:1duIyWiTaYvVx3YX2CYtpJbUFd7/UuPYCfgXtQ3VTbI=
-gopkg.in/jcmturner/goidentity.v3 v3.0.0/go.mod h1:oG2kH0IvSYNIu80dVAyu/yoefjq1mNfM5bm88whjWx4=
-gopkg.in/jcmturner/gokrb5.v7 v7.2.3 h1:hHMV/yKPwMnJhPuPx7pH2Uw/3Qyf+thJYlisUc44010=
-gopkg.in/jcmturner/gokrb5.v7 v7.2.3/go.mod h1:l8VISx+WGYp+Fp7KRbsiUuXTTOnxIc3Tuvyavf11/WM=
-gopkg.in/jcmturner/rpc.v1 v1.1.0 h1:QHIUxTX1ISuAv9dD2wJ9HWQVuWDX/Zc0PfeC2tjc4rU=
-gopkg.in/jcmturner/rpc.v1 v1.1.0/go.mod h1:YIdkC4XfD6GXbzje11McwsDuOlZQSb9W4vfLvuNnlv8=
-gopkg.in/yaml.v2 v2.2.2 h1:ZCJp+EgiOT7lHqUV2J862kp8Qj64Jo6az82+3Td9dZw=
-gopkg.in/yaml.v2 v2.2.2/go.mod h1:hI93XBmqTisBFMUTm0b8Fm+jr3Dg1NNxqwp+5A1VGuI=
--- a/plugins/auth/authfile/aclconfig.go
+++ b/plugins/auth/authfile/aclconfig.go
@@ -1,3 +1,5 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package acl

 import (
@@ -5,13 +7,14 @@ import (
 	"errors"
 	"io"
 	"os"
+	"strconv"
 	"strings"
 )

 const (
-	SUB      = "1"
-	PUB      = "2"
-	PUBSUB   = "3"
+	PUB      = 1
+	SUB      = 2
+	PUBSUB   = 3
 	CLIENTID = "clientid"
 	USERNAME = "username"
 	IP       = "ip"
@@ -23,7 +26,7 @@ type AuthInfo struct {
 	Auth   string
 	Typ    string
 	Val    string
-	PubSub string
+	PubSub int
 	Topics []string
 }

@@ -33,6 +36,9 @@ type ACLConfig struct {
 }

 func AclConfigLoad(file string) (*ACLConfig, error) {
+	if file == "" {
+		file = "./conf/acl.conf"
+	}
 	aclconifg := &ACLConfig{
 		File: file,
 		Info: make([]*AuthInfo, 0, 4),
@@ -75,16 +81,12 @@ func (c *ACLConfig) Prase() error {
 			parseErr = errors.New("\"" + line + "\" format is error")
 			break
 		}
-		if tmpArr[3] != PUB && tmpArr[3] != SUB && tmpArr[3] != PUBSUB {
+		var pubsub int
+		pubsub, err = strconv.Atoi(tmpArr[3])
+		if err != nil {
 			parseErr = errors.New("\"" + line + "\" format is error")
 			break
 		}
-		// var pubsub int
-		// pubsub, err = strconv.Atoi(tmpArr[3])
-		// if err != nil {
-		// 	parseErr = errors.New("\"" + line + "\" format is error")
-		// 	break
-		// }
 		topicStr := strings.Replace(tmpArr[4], " ", "", -1)
 		topicStr = strings.Replace(topicStr, "\n", "", -1)
 		topics := strings.Split(topicStr, ",")
@@ -93,7 +95,7 @@ func (c *ACLConfig) Prase() error {
 			Typ:    tmpArr[1],
 			Val:    tmpArr[2],
 			Topics: topics,
-			PubSub: tmpArr[3],
+			PubSub: pubsub,
 		}
 		c.Info = append(c.Info, tmpAuth)
 		if err != nil {
--- a/plugins/auth/authfile/aclcheck.go
+++ b/plugins/auth/authfile/aclcheck.go
@@ -1,21 +1,22 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>*/
 package acl

 import "strings"

-func checkTopicAuth(ACLInfo *ACLConfig, action, ip, username, clientid, topic string) bool {
+func CheckTopicAuth(ACLInfo *ACLConfig, typ int, ip, username, clientid, topic string) bool {
 	for _, info := range ACLInfo.Info {
 		ctyp := info.Typ
 		switch ctyp {
 		case CLIENTID:
-			if match, auth := info.checkWithClientID(action, clientid, topic); match {
+			if match, auth := info.checkWithClientID(typ, clientid, topic); match {
 				return auth
 			}
 		case USERNAME:
-			if match, auth := info.checkWithUsername(action, username, topic); match {
+			if match, auth := info.checkWithUsername(typ, username, topic); match {
 				return auth
 			}
 		case IP:
-			if match, auth := info.checkWithIP(action, ip, topic); match {
+			if match, auth := info.checkWithIP(typ, ip, topic); match {
 				return auth
 			}
 		}
@@ -23,18 +24,18 @@ func checkTopicAuth(ACLInfo *ACLConfig, action, ip, username, clientid, topic st
 	return false
 }

-func (a *AuthInfo) checkWithClientID(action, clientid, topic string) (bool, bool) {
+func (a *AuthInfo) checkWithClientID(typ int, clientid, topic string) (bool, bool) {
 	auth := false
 	match := false
 	if a.Val == "*" || a.Val == clientid {
 		for _, tp := range a.Topics {
 			des := strings.Replace(tp, "%c", clientid, -1)
-			if action == PUB {
+			if typ == PUB {
 				if pubTopicMatch(topic, des) {
 					match = true
 					auth = a.checkAuth(PUB)
 				}
-			} else if action == SUB {
+			} else if typ == SUB {
 				if subTopicMatch(topic, des) {
 					match = true
 					auth = a.checkAuth(SUB)
@@ -45,18 +46,18 @@ func (a *AuthInfo) checkWithClientID(action, clientid, topic string) (bool, bool
 	return match, auth
 }

-func (a *AuthInfo) checkWithUsername(action, username, topic string) (bool, bool) {
+func (a *AuthInfo) checkWithUsername(typ int, username, topic string) (bool, bool) {
 	auth := false
 	match := false
 	if a.Val == "*" || a.Val == username {
 		for _, tp := range a.Topics {
 			des := strings.Replace(tp, "%u", username, -1)
-			if action == PUB {
+			if typ == PUB {
 				if pubTopicMatch(topic, des) {
 					match = true
 					auth = a.checkAuth(PUB)
 				}
-			} else if action == SUB {
+			} else if typ == SUB {
 				if subTopicMatch(topic, des) {
 					match = true
 					auth = a.checkAuth(SUB)
@@ -67,18 +68,18 @@ func (a *AuthInfo) checkWithUsername(action, username, topic string) (bool, bool
 	return match, auth
 }

-func (a *AuthInfo) checkWithIP(action, ip, topic string) (bool, bool) {
+func (a *AuthInfo) checkWithIP(typ int, ip, topic string) (bool, bool) {
 	auth := false
 	match := false
 	if a.Val == "*" || a.Val == ip {
 		for _, tp := range a.Topics {
 			des := tp
-			if action == PUB {
+			if typ == PUB {
 				if pubTopicMatch(topic, des) {
 					auth = a.checkAuth(PUB)
 					match = true
 				}
-			} else if action == SUB {
+			} else if typ == SUB {
 				if subTopicMatch(topic, des) {
 					auth = a.checkAuth(SUB)
 					match = true
@@ -89,15 +90,15 @@ func (a *AuthInfo) checkWithIP(action, ip, topic string) (bool, bool) {
 	return match, auth
 }

-func (a *AuthInfo) checkAuth(action string) bool {
+func (a *AuthInfo) checkAuth(typ int) bool {
 	auth := false
-	if action == PUB {
+	if typ == PUB {
 		if a.Auth == ALLOW && (a.PubSub == PUB || a.PubSub == PUBSUB) {
 			auth = true
 		} else if a.Auth == DENY && a.PubSub == SUB {
 			auth = true
 		}
-	} else if action == SUB {
+	} else if typ == SUB {
 		if a.Auth == ALLOW && (a.PubSub == SUB || a.PubSub == PUBSUB) {
 			auth = true
 		} else if a.Auth == DENY && a.PubSub == PUB {
--- a/plugins/auth/authfile/spilt.go
+++ b/plugins/auth/authfile/spilt.go
@@ -1,3 +1,5 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package acl

 import (
--- a/broker/lib/sessions/memprovider.go
+++ b/broker/lib/sessions/memprovider.go
@@ -1,3 +1,17 @@
+// Copyright (c) 2014 The SurgeMQ Authors. All rights reserved.
+//
+// Licensed under the Apache License, Version 2.0 (the "License");
+// you may not use this file except in compliance with the License.
+// You may obtain a copy of the License at
+//
+//     http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing, software
+// distributed under the License is distributed on an "AS IS" BASIS,
+// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
+// See the License for the specific language governing permissions and
+// limitations under the License.
+
 package sessions

 import (
--- a/lib/sessions/redisprovider.go
+++ b/lib/sessions/redisprovider.go
@@ -0,0 +1,95 @@
+package sessions
+
+import (
+	"time"
+
+	log "github.com/cihub/seelog"
+	"github.com/go-redis/redis"
+	jsoniter "github.com/json-iterator/go"
+)
+
+var redisClient *redis.Client
+var _ SessionsProvider = (*redisProvider)(nil)
+
+const (
+	sessionName = "session"
+)
+
+type redisProvider struct {
+}
+
+func init() {
+	Register("redis", NewRedisProvider())
+}
+
+func InitRedisConn(url string) {
+	redisClient = redis.NewClient(&redis.Options{
+		Addr:     "127.0.0.1:6379",
+		Password: "", // no password set
+		DB:       0,  // use default DB
+	})
+	err := redisClient.Ping().Err()
+	for err != nil {
+		log.Error("connect redis error: ", err, " 3s try again...")
+		time.Sleep(3 * time.Second)
+		err = redisClient.Ping().Err()
+	}
+}
+
+func NewRedisProvider() *redisProvider {
+	return &redisProvider{}
+}
+
+func (r *redisProvider) New(id string) (*Session, error) {
+	val, _ := jsoniter.Marshal(&Session{id: id})
+
+	err := redisClient.HSet(sessionName, id, val).Err()
+	if err != nil {
+		return nil, err
+	}
+
+	result, err := redisClient.HGet(sessionName, id).Bytes()
+	if err != nil {
+		return nil, err
+	}
+
+	sess := Session{}
+	err = jsoniter.Unmarshal(result, &sess)
+	if err != nil {
+		return nil, err
+	}
+
+	return &sess, nil
+}
+
+func (r *redisProvider) Get(id string) (*Session, error) {
+
+	result, err := redisClient.HGet(sessionName, id).Bytes()
+	if err != nil {
+		return nil, err
+	}
+
+	sess := Session{}
+	err = jsoniter.Unmarshal(result, &sess)
+	if err != nil {
+		return nil, err
+	}
+
+	return &sess, nil
+}
+
+func (r *redisProvider) Del(id string) {
+	redisClient.HDel(sessionName, id)
+}
+
+func (r *redisProvider) Save(id string) error {
+	return nil
+}
+
+func (r *redisProvider) Count() int {
+	return int(redisClient.HLen(sessionName).Val())
+}
+
+func (r *redisProvider) Close() error {
+	return redisClient.Del(sessionName).Err()
+}
--- a/broker/lib/sessions/session.go
+++ b/broker/lib/sessions/session.go
--- a/broker/lib/sessions/sessions.go
+++ b/broker/lib/sessions/sessions.go
--- a/broker/lib/topics/memtopics.go
+++ b/broker/lib/topics/memtopics.go
@@ -104,7 +104,7 @@ func (this *memTopics) Retain(msg *packets.PublishPacket) error {
 		return this.rroot.rremove([]byte(msg.TopicName))
 	}

-	return this.rroot.rinsertOrUpdate([]byte(msg.TopicName), msg)
+	return this.rroot.rinsert([]byte(msg.TopicName), msg)
 }

 func (this *memTopics) Retained(topic []byte, msgs *[]*packets.PublishPacket) error {
@@ -244,9 +244,6 @@ func (this *snode) smatch(topic []byte, qos byte, subs *[]interface{}, qoss *[]b
 	// let's find the subscribers that match the qos and append them to the list.
 	if len(topic) == 0 {
 		this.matchQos(qos, subs, qoss)
-		if mwcn, _ := this.snodes[MWC]; mwcn != nil {
-			mwcn.matchQos(qos, subs, qoss)
-		}
 		return nil
 	}

@@ -286,11 +283,13 @@ func newRNode() *rnode {
 	}
 }

-func (this *rnode) rinsertOrUpdate(topic []byte, msg *packets.PublishPacket) error {
+func (this *rnode) rinsert(topic []byte, msg *packets.PublishPacket) error {
 	// If there's no more topic levels, that means we are at the matching rnode.
 	if len(topic) == 0 {
 		// Reuse the message if possible
-		this.msg = msg
+		if this.msg == nil {
+			this.msg = msg
+		}

 		return nil
 	}
@@ -313,7 +312,7 @@ func (this *rnode) rinsertOrUpdate(topic []byte, msg *packets.PublishPacket) err
 		this.rnodes[level] = n
 	}

-	return n.rinsertOrUpdate(rem, msg)
+	return n.rinsert(rem, msg)
 }

 // Remove the retained message for the supplied topic
--- a/broker/lib/topics/topics.go
+++ b/broker/lib/topics/topics.go
--- a/logger/logger.go
+++ b/logger/logger.go
@@ -5,27 +5,17 @@ package logger

 import (
 	"go.uber.org/zap"
-	"go.uber.org/zap/zapcore"
 )

 var (
 	// env can be setup at build time with Go Linker. Value could be prod or whatever else for dev env
-	instance   *zap.Logger
-	logCfg     zap.Config
-	encoderCfg = zap.NewProductionEncoderConfig()
+	instance *zap.Logger
+	logCfg   zap.Config
 )

-func init() {
-	encoderCfg.TimeKey = "timestamp"
-	encoderCfg.EncodeTime = zapcore.ISO8601TimeEncoder
-}
-
 // NewDevLogger return a logger for dev builds
 func NewDevLogger() (*zap.Logger, error) {
-	logCfg := zap.NewProductionConfig()
-	logCfg.Level = zap.NewAtomicLevelAt(zap.DebugLevel)
-	// logCfg.DisableStacktrace = true
-	logCfg.EncoderConfig = encoderCfg
+	logCfg := zap.NewDevelopmentConfig()
 	return logCfg.Build()
 }

@@ -34,31 +24,27 @@ func NewProdLogger() (*zap.Logger, error) {
 	logCfg := zap.NewProductionConfig()
 	logCfg.DisableStacktrace = true
 	logCfg.Level = zap.NewAtomicLevelAt(zap.InfoLevel)
-	logCfg.EncoderConfig = encoderCfg
 	return logCfg.Build()
 }

-func Prod() *zap.Logger {
-
-	l, _ := NewProdLogger()
-	instance = l
-
-	return instance
-}
-
-func Debug() *zap.Logger {
-
-	l, _ := NewDevLogger()
-	instance = l
-
-	return instance
-}
-
-func Get() *zap.Logger {
-	if instance == nil {
-		l, _ := NewProdLogger()
-		instance = l
+func InitLogger(debug bool) {
+	var err error
+	var log *zap.Logger
+	if debug {
+		log, err = NewDevLogger()
+	} else {
+		log, err = NewProdLogger()
 	}
+	if err != nil {
+		panic("Unable to create a logger.")
+	}
+	defer log.Sync()

+	log.Debug("Logger initialization succeeded")
+	instance = log.Named("hmq")
+}
+
+// Get return a *zap.Logger instance
+func Get() *zap.Logger {
 	return instance
 }
--- a/logger/logger_test.go
+++ b/logger/logger_test.go
@@ -1,6 +1,5 @@
-/*
-Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
-*/
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+ */
 package logger

 import (
--- a/main.go
+++ b/main.go
@@ -1,7 +1,13 @@
+/* Copyright (c) 2018, joy.zhou <chowyu08@gmail.com>
+
+Permission to use, copy, modify, and/or distribute this software for any
+purpose with or without fee is hereby granted, provided that the above
+copyright notice and this permission notice appear in all copies.
+*/
 package main

 import (
-	"log"
+	"fmt"
 	"os"
 	"os/signal"
 	"runtime"
@@ -13,17 +19,19 @@ func main() {
 	runtime.GOMAXPROCS(runtime.NumCPU())
 	config, err := broker.ConfigureConfig(os.Args[1:])
 	if err != nil {
-		log.Fatal("configure broker config error: ", err)
+		fmt.Println("configure broker config error: ", err)
+		return
 	}

 	b, err := broker.NewBroker(config)
 	if err != nil {
-		log.Fatal("New Broker error: ", err)
+		fmt.Println("New Broker error: ", err)
+		return
 	}
 	b.Start()

 	s := waitForSignal()
-	log.Println("signal received, broker closed.", s)
+	fmt.Println("signal received, broker closed.", s)
 }

 func waitForSignal() os.Signal {
--- a/plugins/auth/auth.go
+++ b/plugins/auth/auth.go
@@ -1,27 +0,0 @@
-package auth
-
-import (
-	authfile "github.com/fhmq/hmq/plugins/auth/authfile"
-	"github.com/fhmq/hmq/plugins/auth/authhttp"
-)
-
-const (
-	AuthHTTP = "authhttp"
-	AuthFile = "authfile"
-)
-
-type Auth interface {
-	CheckACL(action, clientID, username, ip, topic string) bool
-	CheckConnect(clientID, username, password string) bool
-}
-
-func NewAuth(name string) Auth {
-	switch name {
-	case AuthHTTP:
-		return authhttp.Init()
-	case AuthFile:
-		return authfile.Init()
-	default:
-		return &mockAuth{}
-	}
-}
--- a/plugins/auth/authfile/Readme.md
+++ b/plugins/auth/authfile/Readme.md
@@ -1,54 +0,0 @@
-## ACL Configure
-```
-Attention: Acl Type Change, change `pub =1, sub=2`  to `sub =1, pub=2`
-```
-#### The ACL rules define:
-~~~
-Allow | type | value | pubsub | Topics
-~~~
-#### ACL Config
-~~~
-## type clientid , username, ipaddr
-##sub 1 ,  pub 2,  pubsub 3
-## %c is clientid , %u is username
-allow      ip          127.0.0.1   2     $SYS/#
-allow      clientid    0001        3     #
-allow      username    admin       3     #
-allow      username    joy         3     /test,hello/world 
-allow      clientid    *           1     toCloud/%c
-allow      username    *           1     toCloud/%u
-deny       clientid    *           3     #
-~~~
-
-~~~
-#allow local sub $SYS topic
-allow      ip          127.0.0.1   1    $SYS/#
-~~~
-~~~
-#allow client who's id with 0001 or username with admin pub sub all topic
-allow      clientid    0001        3        #
-allow      username    admin       3        #
-~~~
-~~~
-#allow client with the username joy can pub sub topic '/test' and 'hello/world'
-allow      username    joy         3     /test,hello/world 
-~~~
-~~~
-#allow all client pub the topic toCloud/{clientid/username}
-allow      clientid    *         2         toCloud/%c
-allow      username    *         2         toCloud/%u
-~~~
-~~~
-#deny all client pub sub all topic
-deny       clientid    *         3           #
-~~~
-Client match acl rule one by one
-~~~
-          ---------              ---------              ---------
-Client -> | Rule1 | --nomatch--> | Rule2 | --nomatch--> | Rule3 | --> 
-          ---------              ---------              ---------
-              |                      |                      |
-            match                  match                  match
-             \|/                    \|/                    \|/
-        allow | deny           allow | deny           allow | deny
-~~~
--- a/plugins/auth/authfile/acl.go
+++ b/plugins/auth/authfile/acl.go
@@ -1,23 +0,0 @@
-package acl
-
-type aclAuth struct {
-	config *ACLConfig
-}
-
-func Init() *aclAuth {
-	aclConfig, err := AclConfigLoad("./plugins/auth/authfile/acl.conf")
-	if err != nil {
-		panic(err)
-	}
-	return &aclAuth{
-		config: aclConfig,
-	}
-}
-
-func (a *aclAuth) CheckConnect(clientID, username, password string) bool {
-	return true
-}
-
-func (a *aclAuth) CheckACL(action, clientID, username, ip, topic string) bool {
-	return checkTopicAuth(a.config, action, username, ip, clientID, topic)
-}
--- a/plugins/auth/authhttp/authhttp.go
+++ b/plugins/auth/authhttp/authhttp.go
@@ -1,179 +0,0 @@
-package authhttp
-
-import (
-	"encoding/json"
-	"io"
-	"io/ioutil"
-	"net/http"
-	"net/url"
-	"strconv"
-	"strings"
-	"time"
-
-	"github.com/fhmq/hmq/logger"
-	"go.uber.org/zap"
-)
-
-//Config device kafka config
-type Config struct {
-	AuthURL  string `json:"auth"`
-	ACLURL   string `json:"acl"`
-	SuperURL string `json:"super"`
-}
-
-type authHTTP struct {
-	client *http.Client
-}
-
-var (
-	config     Config
-	log        = logger.Get().Named("authhttp")
-	httpClient *http.Client
-)
-
-//Init init kafak client
-func Init() *authHTTP {
-	content, err := ioutil.ReadFile("./plugins/auth/authhttp/http.json")
-	if err != nil {
-		log.Fatal("Read config file error: ", zap.Error(err))
-	}
-	// log.Info(string(content))
-
-	err = json.Unmarshal(content, &config)
-	if err != nil {
-		log.Fatal("Unmarshal config file error: ", zap.Error(err))
-	}
-	// fmt.Println("http: config: ", config)
-
-	httpClient = &http.Client{
-		Transport: &http.Transport{
-			MaxConnsPerHost:     100,
-			MaxIdleConns:        100,
-			MaxIdleConnsPerHost: 100,
-		},
-		Timeout: time.Second * 100,
-	}
-	return &authHTTP{client: httpClient}
-}
-
-//CheckAuth check mqtt connect
-func (a *authHTTP) CheckConnect(clientID, username, password string) bool {
-	action := "connect"
-	{
-		aCache := checkCache(action, clientID, username, password, "")
-		if aCache != nil {
-			if aCache.password == password && aCache.username == username && aCache.action == action {
-				return true
-			}
-		}
-	}
-
-	data := url.Values{}
-	data.Add("username", username)
-	data.Add("clientid", clientID)
-	data.Add("password", password)
-
-	req, err := http.NewRequest("POST", config.AuthURL, strings.NewReader(data.Encode()))
-	if err != nil {
-		log.Error("new request super: ", zap.Error(err))
-		return false
-	}
-	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
-	req.Header.Add("Content-Length", strconv.Itoa(len(data.Encode())))
-
-	resp, err := a.client.Do(req)
-	if err != nil {
-		log.Error("request super: ", zap.Error(err))
-		return false
-	}
-
-	defer resp.Body.Close()
-	io.Copy(ioutil.Discard, resp.Body)
-	if resp.StatusCode == http.StatusOK {
-		addCache(action, clientID, username, password, "")
-		return true
-	}
-
-	return false
-}
-
-// //CheckSuper check mqtt connect
-// func CheckSuper(clientID, username, password string) bool {
-// 	action := "connect"
-// 	{
-// 		aCache := checkCache(action, clientID, username, password, "")
-// 		if aCache != nil {
-// 			if aCache.password == password && aCache.username == username && aCache.action == action {
-// 				return true
-// 			}
-// 		}
-// 	}
-
-// 	data := url.Values{}
-// 	data.Add("username", username)
-// 	data.Add("clientid", clientID)
-// 	data.Add("password", password)
-
-// 	req, err := http.NewRequest("POST", config.SuperURL, strings.NewReader(data.Encode()))
-// 	if err != nil {
-// 		log.Error("new request super: ", zap.Error(err))
-// 		return false
-// 	}
-// 	req.Header.Add("Content-Type", "application/x-www-form-urlencoded")
-// 	req.Header.Add("Content-Length", strconv.Itoa(len(data.Encode())))
-
-// 	resp, err := httpClient.Do(req)
-// 	if err != nil {
-// 		log.Error("request super: ", zap.Error(err))
-// 		return false
-// 	}
-
-// 	defer resp.Body.Close()
-// 	io.Copy(ioutil.Discard, resp.Body)
-
-// 	if resp.StatusCode == http.StatusOK {
-// 		return true
-// 	}
-// 	return false
-// }
-
-//CheckACL check mqtt connect
-func (a *authHTTP) CheckACL(action, clientID, username, ip, topic string) bool {
-
-	{
-		aCache := checkCache(action, "", username, "", topic)
-		if aCache != nil {
-			if aCache.topic == topic && aCache.action == action {
-				return true
-			}
-		}
-	}
-
-	req, err := http.NewRequest("GET", config.ACLURL, nil)
-	if err != nil {
-		log.Error("get acl: ", zap.Error(err))
-		return false
-	}
-
-	data := req.URL.Query()
-
-	data.Add("username", username)
-	data.Add("topic", topic)
-	data.Add("access", action)
-	req.URL.RawQuery = data.Encode()
-	// fmt.Println("req:", req)
-	resp, err := a.client.Do(req)
-	if err != nil {
-		log.Error("request acl: ", zap.Error(err))
-		return false
-	}
-
-	defer resp.Body.Close()
-	io.Copy(ioutil.Discard, resp.Body)
-
-	if resp.StatusCode == http.StatusOK {
-		addCache(action, "", username, "", topic)
-		return true
-	}
-	return false
-}
--- a/plugins/auth/authhttp/cache.go
+++ b/plugins/auth/authhttp/cache.go
@@ -1,32 +0,0 @@
-package authhttp
-
-import (
-	"time"
-
-	"github.com/patrickmn/go-cache"
-)
-
-type authCache struct {
-	action   string
-	username string
-	clientID string
-	password string
-	topic    string
-}
-
-var (
-	// cache = make(map[string]authCache)
-	c = cache.New(5*time.Minute, 10*time.Minute)
-)
-
-func checkCache(action, clientID, username, password, topic string) *authCache {
-	authc, found := c.Get(username)
-	if found {
-		return authc.(*authCache)
-	}
-	return nil
-}
-
-func addCache(action, clientID, username, password, topic string) {
-	c.Set(username, &authCache{action: action, username: username, clientID: clientID, password: password, topic: topic}, cache.DefaultExpiration)
-}
--- a/plugins/auth/authhttp/http.json
+++ b/plugins/auth/authhttp/http.json
@@ -1,5 +0,0 @@
-{
-    "auth": "http://127.0.0.1:9090/mqtt/auth",
-    "acl": "http://127.0.0.1:9090/mqtt/acl",
-    "super": "http://127.0.0.1:9090/mqtt/superuser"
-}
--- a/plugins/auth/mock.go
+++ b/plugins/auth/mock.go
@@ -1,11 +0,0 @@
-package auth
-
-type mockAuth struct{}
-
-func (m *mockAuth) CheckACL(action, clientID, username, ip, topic string) bool {
-	return true
-}
-
-func (m *mockAuth) CheckConnect(clientID, username, password string) bool {
-	return true
-}
--- a/plugins/bridge/bridge.go
+++ b/plugins/bridge/bridge.go
@@ -1,49 +0,0 @@
-package bridge
-
-import "github.com/fhmq/hmq/logger"
-
-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"
-)
-
-var (
-	log = logger.Get().Named("bridge")
-)
-
-//Elements kafka publish elements
-type Elements struct {
-	ClientID  string `json:"clientid"`
-	Username  string `json:"username"`
-	Topic     string `json:"topic"`
-	Payload   string `json:"payload"`
-	Timestamp int64  `json:"ts"`
-	Size      int32  `json:"size"`
-	Action    string `json:"action"`
-}
-
-const (
-	//Kafka plugin name
-	Kafka = "kafka"
-)
-
-type BridgeMQ interface {
-	Publish(e *Elements) error
-}
-
-func NewBridgeMQ(name string) BridgeMQ {
-	switch name {
-	case Kafka:
-		return InitKafka()
-	default:
-		return &mockMQ{}
-	}
-}
--- a/plugins/bridge/kafka.go
+++ b/plugins/bridge/kafka.go
@@ -1,156 +0,0 @@
-package bridge
-
-import (
-	"encoding/json"
-	"errors"
-	"io/ioutil"
-	"strings"
-	"time"
-
-	"github.com/Shopify/sarama"
-	"go.uber.org/zap"
-)
-
-type kafakConfig struct {
-	Addr             []string          `json:"addr"`
-	ConnectTopic     string            `json:"onConnect"`
-	SubscribeTopic   string            `json:"onSubscribe"`
-	PublishTopic     string            `json:"onPublish"`
-	UnsubscribeTopic string            `json:"onUnsubscribe"`
-	DisconnectTopic  string            `json:"onDisconnect"`
-	DeliverMap       map[string]string `json:"deliverMap"`
-}
-
-type kafka struct {
-	kafakConfig kafakConfig
-	kafkaClient sarama.AsyncProducer
-}
-
-//Init init kafak client
-func InitKafka() *kafka {
-	log.Info("start connect kafka....")
-	content, err := ioutil.ReadFile("./plugins/kafka/kafka.json")
-	if err != nil {
-		log.Fatal("Read config file error: ", zap.Error(err))
-	}
-	// log.Info(string(content))
-	var config kafakConfig
-	err = json.Unmarshal(content, &config)
-	if err != nil {
-		log.Fatal("Unmarshal config file error: ", zap.Error(err))
-	}
-	c := &kafka{kafakConfig: config}
-	c.connect()
-	return c
-}
-
-//connect
-func (k *kafka) connect() {
-	conf := sarama.NewConfig()
-	conf.Version = sarama.V1_1_1_0
-	kafkaClient, err := sarama.NewAsyncProducer(k.kafakConfig.Addr, conf)
-	if err != nil {
-		log.Fatal("create kafka async producer failed: ", zap.Error(err))
-	}
-
-	go func() {
-		for err := range kafkaClient.Errors() {
-			log.Error("send msg to kafka failed: ", zap.Error(err))
-		}
-	}()
-
-	k.kafkaClient = kafkaClient
-}
-
-//Publish publish to kafka
-func (k *kafka) Publish(e *Elements) error {
-	config := k.kafakConfig
-	key := e.ClientID
-	topics := make(map[string]bool)
-	switch e.Action {
-	case Connect:
-		if config.ConnectTopic != "" {
-			topics[config.ConnectTopic] = true
-		}
-	case Publish:
-		if config.PublishTopic != "" {
-			topics[config.PublishTopic] = true
-		}
-		// foreach regexp map config
-		for reg, topic := range config.DeliverMap {
-			match := matchTopic(reg, e.Topic)
-			if match {
-				topics[topic] = true
-			}
-		}
-	case Subscribe:
-		if config.SubscribeTopic != "" {
-			topics[config.SubscribeTopic] = true
-		}
-	case Unsubscribe:
-		if config.UnsubscribeTopic != "" {
-			topics[config.UnsubscribeTopic] = true
-		}
-	case Disconnect:
-		if config.DisconnectTopic != "" {
-			topics[config.DisconnectTopic] = true
-		}
-	default:
-		return errors.New("error action: " + e.Action)
-	}
-
-	return k.publish(topics, key, e)
-
-}
-
-func (k *kafka) publish(topics map[string]bool, key string, msg *Elements) error {
-	payload, err := json.Marshal(msg)
-	if err != nil {
-		return err
-	}
-
-	for topic, _ := range topics {
-		select {
-		case k.kafkaClient.Input() <- &sarama.ProducerMessage{
-			Topic: topic,
-			Key:   sarama.ByteEncoder(key),
-			Value: sarama.ByteEncoder(payload),
-		}:
-			continue
-		case <-time.After(5 * time.Second):
-			return errors.New("write kafka timeout")
-		}
-
-	}
-
-	return nil
-}
-
-func match(subTopic []string, topic []string) bool {
-	if len(subTopic) == 0 {
-		if len(topic) == 0 {
-			return true
-		}
-		return false
-	}
-
-	if len(topic) == 0 {
-		if subTopic[0] == "#" {
-			return true
-		}
-		return false
-	}
-
-	if subTopic[0] == "#" {
-		return true
-	}
-
-	if (subTopic[0] == "+") || (subTopic[0] == topic[0]) {
-		return match(subTopic[1:], topic[1:])
-	}
-	return false
-}
-
-func matchTopic(subTopic string, topic string) bool {
-	return match(strings.Split(subTopic, "/"), strings.Split(topic, "/"))
-}
--- a/plugins/bridge/kafka/kafka.json
+++ b/plugins/bridge/kafka/kafka.json
@@ -1,14 +0,0 @@
-{
-    "addr": [
-        "127.0.0.1:9090"
-    ],
-    "onConnect": "onConnect",
-    "onPublish": "onPublish",
-    "onSubscribe": "onSubscribe",
-    "onDisconnect": "onDisconnect",
-    "onUnsubscribe": "onUnsubscribe",
-    "deliverMap": {
-        "#": "publish",
-        "/upload/+/#": "upload"
-    }
-}
--- a/plugins/bridge/mock.go
+++ b/plugins/bridge/mock.go
@@ -1,7 +0,0 @@
-package bridge
-
-type mockMQ struct{}
-
-func (m *mockMQ) Publish(e *Elements) error {
-	return nil
-}
--- a/pool/fixpool.go
+++ b/pool/fixpool.go
@@ -1,58 +0,0 @@
-package pool
-
-import (
-	"github.com/segmentio/fasthash/fnv1a"
-)
-
-type WorkerPool struct {
-	maxWorkers  int
-	taskQueue   []chan func()
-	stoppedChan chan struct{}
-}
-
-func New(maxWorkers int) *WorkerPool {
-	// There must be at least one worker.
-	if maxWorkers < 1 {
-		maxWorkers = 1
-	}
-
-	// taskQueue is unbuffered since items are always removed immediately.
-	pool := &WorkerPool{
-		taskQueue:   make([]chan func(), maxWorkers),
-		maxWorkers:  maxWorkers,
-		stoppedChan: make(chan struct{}),
-	}
-	// Start the task dispatcher.
-	pool.dispatch()
-
-	return pool
-}
-
-func (p *WorkerPool) Submit(uid string, task func()) {
-	idx := fnv1a.HashString64(uid) % uint64(p.maxWorkers)
-	if task != nil {
-		p.taskQueue[idx] <- task
-	}
-}
-
-func (p *WorkerPool) dispatch() {
-	for i := 0; i < p.maxWorkers; i++ {
-		p.taskQueue[i] = make(chan func(), 1024)
-		go startWorker(p.taskQueue[i])
-	}
-}
-
-func startWorker(taskChan chan func()) {
-	go func() {
-		var task func()
-		var ok bool
-		for {
-			task, ok = <-taskChan
-			if !ok {
-				break
-			}
-			// Execute the task.
-			task()
-		}
-	}()
-}
--- a/pool/pool.go
+++ b/pool/pool.go
@@ -1,166 +1,166 @@
 package pool

-// import "time"
+import "time"

-// const (
-// 	// This value is the size of the queue that workers register their
-// 	// availability to the dispatcher.  There may be hundreds of workers, but
-// 	// only a small channel is needed to register some of the workers.
-// 	readyQueueSize = 64
+const (
+	// This value is the size of the queue that workers register their
+	// availability to the dispatcher.  There may be hundreds of workers, but
+	// only a small channel is needed to register some of the workers.
+	readyQueueSize = 16

-// 	// If worker pool receives no new work for this period of time, then stop
-// 	// a worker goroutine.
-// 	idleTimeoutSec = 5
-// )
+	// If worker pool receives no new work for this period of time, then stop
+	// a worker goroutine.
+	idleTimeoutSec = 5
+)

-// type WorkerPool struct {
-// 	maxWorkers   int
-// 	timeout      time.Duration
-// 	taskQueue    chan func()
-// 	readyWorkers chan chan func()
-// 	stoppedChan  chan struct{}
-// }
+type WorkerPool struct {
+	maxWorkers   int
+	timeout      time.Duration
+	taskQueue    chan func()
+	readyWorkers chan chan func()
+	stoppedChan  chan struct{}
+}

-// func New(maxWorkers int) *WorkerPool {
-// 	// There must be at least one worker.
-// 	if maxWorkers < 1 {
-// 		maxWorkers = 1
-// 	}
+func New(maxWorkers int) *WorkerPool {
+	// There must be at least one worker.
+	if maxWorkers < 1 {
+		maxWorkers = 1
+	}

-// 	// taskQueue is unbuffered since items are always removed immediately.
-// 	pool := &WorkerPool{
-// 		taskQueue:    make(chan func()),
-// 		maxWorkers:   maxWorkers,
-// 		readyWorkers: make(chan chan func(), readyQueueSize),
-// 		timeout:      time.Second * idleTimeoutSec,
-// 		stoppedChan:  make(chan struct{}),
-// 	}
+	// taskQueue is unbuffered since items are always removed immediately.
+	pool := &WorkerPool{
+		taskQueue:    make(chan func()),
+		maxWorkers:   maxWorkers,
+		readyWorkers: make(chan chan func(), readyQueueSize),
+		timeout:      time.Second * idleTimeoutSec,
+		stoppedChan:  make(chan struct{}),
+	}

-// 	// Start the task dispatcher.
-// 	go pool.dispatch()
+	// Start the task dispatcher.
+	go pool.dispatch()

-// 	return pool
-// }
+	return pool
+}

-// func (p *WorkerPool) Stop() {
-// 	if p.Stopped() {
-// 		return
-// 	}
-// 	close(p.taskQueue)
-// 	<-p.stoppedChan
-// }
+func (p *WorkerPool) Stop() {
+	if p.Stopped() {
+		return
+	}
+	close(p.taskQueue)
+	<-p.stoppedChan
+}

-// func (p *WorkerPool) Stopped() bool {
-// 	select {
-// 	case <-p.stoppedChan:
-// 		return true
-// 	default:
-// 	}
-// 	return false
-// }
+func (p *WorkerPool) Stopped() bool {
+	select {
+	case <-p.stoppedChan:
+		return true
+	default:
+	}
+	return false
+}

-// func (p *WorkerPool) Submit(task func()) {
-// 	if task != nil {
-// 		p.taskQueue <- task
-// 	}
-// }
+func (p *WorkerPool) Submit(task func()) {
+	if task != nil {
+		p.taskQueue <- task
+	}
+}

-// func (p *WorkerPool) SubmitWait(task func()) {
-// 	if task == nil {
-// 		return
-// 	}
-// 	doneChan := make(chan struct{})
-// 	p.taskQueue <- func() {
-// 		task()
-// 		close(doneChan)
-// 	}
-// 	<-doneChan
-// }
+func (p *WorkerPool) SubmitWait(task func()) {
+	if task == nil {
+		return
+	}
+	doneChan := make(chan struct{})
+	p.taskQueue <- func() {
+		task()
+		close(doneChan)
+	}
+	<-doneChan
+}

-// func (p *WorkerPool) dispatch() {
-// 	defer close(p.stoppedChan)
-// 	timeout := time.NewTimer(p.timeout)
-// 	var workerCount int
-// 	var task func()
-// 	var ok bool
-// 	var workerTaskChan chan func()
-// 	startReady := make(chan chan func())
-// Loop:
-// 	for {
-// 		timeout.Reset(p.timeout)
-// 		select {
-// 		case task, ok = <-p.taskQueue:
-// 			if !ok {
-// 				break Loop
-// 			}
-// 			// Got a task to do.
-// 			select {
-// 			case workerTaskChan = <-p.readyWorkers:
-// 				// A worker is ready, so give task to worker.
-// 				workerTaskChan <- task
-// 			default:
-// 				// No workers ready.
-// 				// Create a new worker, if not at max.
-// 				if workerCount < p.maxWorkers {
-// 					workerCount++
-// 					go func(t func()) {
-// 						startWorker(startReady, p.readyWorkers)
-// 						// Submit the task when the new worker.
-// 						taskChan := <-startReady
-// 						taskChan <- t
-// 					}(task)
-// 				} else {
-// 					// Start a goroutine to submit the task when an existing
-// 					// worker is ready.
-// 					go func(t func()) {
-// 						taskChan := <-p.readyWorkers
-// 						taskChan <- t
-// 					}(task)
-// 				}
-// 			}
-// 		case <-timeout.C:
-// 			// Timed out waiting for work to arrive.  Kill a ready worker.
-// 			if workerCount > 0 {
-// 				select {
-// 				case workerTaskChan = <-p.readyWorkers:
-// 					// A worker is ready, so kill.
-// 					close(workerTaskChan)
-// 					workerCount--
-// 				default:
-// 					// No work, but no ready workers.  All workers are busy.
-// 				}
-// 			}
-// 		}
-// 	}
+func (p *WorkerPool) dispatch() {
+	defer close(p.stoppedChan)
+	timeout := time.NewTimer(p.timeout)
+	var workerCount int
+	var task func()
+	var ok bool
+	var workerTaskChan chan func()
+	startReady := make(chan chan func())
+Loop:
+	for {
+		timeout.Reset(p.timeout)
+		select {
+		case task, ok = <-p.taskQueue:
+			if !ok {
+				break Loop
+			}
+			// Got a task to do.
+			select {
+			case workerTaskChan = <-p.readyWorkers:
+				// A worker is ready, so give task to worker.
+				workerTaskChan <- task
+			default:
+				// No workers ready.
+				// Create a new worker, if not at max.
+				if workerCount < p.maxWorkers {
+					workerCount++
+					go func(t func()) {
+						startWorker(startReady, p.readyWorkers)
+						// Submit the task when the new worker.
+						taskChan := <-startReady
+						taskChan <- t
+					}(task)
+				} else {
+					// Start a goroutine to submit the task when an existing
+					// worker is ready.
+					go func(t func()) {
+						taskChan := <-p.readyWorkers
+						taskChan <- t
+					}(task)
+				}
+			}
+		case <-timeout.C:
+			// Timed out waiting for work to arrive.  Kill a ready worker.
+			if workerCount > 0 {
+				select {
+				case workerTaskChan = <-p.readyWorkers:
+					// A worker is ready, so kill.
+					close(workerTaskChan)
+					workerCount--
+				default:
+					// No work, but no ready workers.  All workers are busy.
+				}
+			}
+		}
+	}

-// 	// Stop all remaining workers as they become ready.
-// 	for workerCount > 0 {
-// 		workerTaskChan = <-p.readyWorkers
-// 		close(workerTaskChan)
-// 		workerCount--
-// 	}
-// }
+	// Stop all remaining workers as they become ready.
+	for workerCount > 0 {
+		workerTaskChan = <-p.readyWorkers
+		close(workerTaskChan)
+		workerCount--
+	}
+}

-// func startWorker(startReady, readyWorkers chan chan func()) {
-// 	go func() {
-// 		taskChan := make(chan func())
-// 		var task func()
-// 		var ok bool
-// 		// Register availability on starReady channel.
-// 		startReady <- taskChan
-// 		for {
-// 			// Read task from dispatcher.
-// 			task, ok = <-taskChan
-// 			if !ok {
-// 				// Dispatcher has told worker to stop.
-// 				break
-// 			}
+func startWorker(startReady, readyWorkers chan chan func()) {
+	go func() {
+		taskChan := make(chan func())
+		var task func()
+		var ok bool
+		// Register availability on starReady channel.
+		startReady <- taskChan
+		for {
+			// Read task from dispatcher.
+			task, ok = <-taskChan
+			if !ok {
+				// Dispatcher has told worker to stop.
+				break
+			}

-// 			// Execute the task.
-// 			task()
+			// Execute the task.
+			task()

-// 			// Register availability on readyWorkers channel.
-// 			readyWorkers <- taskChan
-// 		}
-// 	}()
-// }
+			// Register availability on readyWorkers channel.
+			readyWorkers <- taskChan
+		}
+	}()
+}
Author	SHA1	Message	Date
Marc Magnin	cf77eaf346	remove subscriptions when a client disconnect	2019-01-18 14:38:51 +01:00
Marc Magnin	7c4d7a0c06	simple fix	2019-01-03 21:23:14 +01:00