add filter function on message process

service/process.go

@@ -101,7 +101,12 @@ func (this *service) processor() {
 
 func (this *service) processIncoming(msg message.Message) error {
 	var err error = nil
-
+	if this.messageFilter != nil {
+		err = this.messageFilter(this.sess.Cmsg, msg)
+		if err != nil {
+			return err
+		}
+	}
 	switch msg := msg.(type) {
 	case *message.PublishMessage:
 		// For PUBLISH message, we should figure out what QoS it is and process accordingly

service/server.go

@@ -48,6 +48,10 @@ const (
 	DefaultTopicsProvider   = "mem"
 )
 
+type (
+	MessageFilterFunc func(cmsg *message.ConnectMessage, msg message.Message) error
+)
+
 // Server is a library implementation of the MQTT server that, as best it can, complies
 // with the MQTT 3.1 and 3.1.1 specs.
 type Server struct {
@@ -109,8 +113,9 @@ type Server struct {
 	// A indicator on whether this server has already checked configuration
 	configOnce sync.Once
 
-	subs []interface{}
-	qoss []byte
+	subs          []interface{}
+	qoss          []byte
+	MessageFilter MessageFilterFunc
 }
 
 // ListenAndServe listents to connections on the URI requested, and handles any
@@ -310,9 +315,10 @@ func (this *Server) handleConnection(c io.Closer) (svc *service, err error) {
 		ackTimeout:     this.AckTimeout,
 		timeoutRetries: this.TimeoutRetries,
 
-		conn:      conn,
-		sessMgr:   this.sessMgr,
-		topicsMgr: this.topicsMgr,
+		conn:          conn,
+		sessMgr:       this.sessMgr,
+		topicsMgr:     this.topicsMgr,
+		messageFilter: this.MessageFilter,
 	}
 
 	err = this.getSession(svc, req, resp)

service/service.go

@@ -118,9 +118,10 @@ type service struct {
 	intmp  []byte
 	outtmp []byte
 
-	subs  []interface{}
-	qoss  []byte
-	rmsgs []*message.PublishMessage
+	subs          []interface{}
+	qoss          []byte
+	rmsgs         []*message.PublishMessage
+	messageFilter MessageFilterFunc
 }
 
 func (this *service) start() error {